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

Merge pull request 'feature/wait-until-the-right-time-before-booking' (#6) from feature/wait-until-the-right-time-before-booking into main

Browse source 
Reviewed-on: https://jouf.fr/gitea/stanislas/resa-padel/pulls/6
This commit is contained in:
Stanislas Jouffroy 2024-02-17 20:42:43 +00:00
commit 78147c5c82
9 changed files with 426 additions and 170 deletions
Download patch file Download diff file Expand all files Collapse all files

66
resa_padel/booking.py
View file

@ -1,33 +1,63 @@
import asyncio
import logging
import time
import pendulum
from aiohttp import ClientSession
from pendulum import DateTime
import config
from aiohttp import ClientSession
from gestion_sports.gestion_sports_connector import GestionSportsConnector
from models import BookingFilter, User
from models import BookingFilter, Club, User
LOGGER = logging.getLogger(__name__)
async def book(url: str, user: User, booking_filter: BookingFilter) -> None:
def wait_until_booking_time(club: Club, booking_filter: BookingFilter):
LOGGER.info("Waiting booking time")
booking_datetime = build_booking_datetime(booking_filter, club)
now = pendulum.now()
while now < booking_datetime:
time.sleep(1)
now = pendulum.now()
def build_booking_datetime(booking_filter: BookingFilter, club: Club) -> DateTime:
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)
async def book(club: Club, user: User, booking_filter: BookingFilter) -> int | None:
async with ClientSession() as session:
platform = GestionSportsConnector(session, url)
platform = GestionSportsConnector(session, club.url)
await platform.connect()
await platform.login(user)
await platform.book(booking_filter)
await platform.login(user, club)
wait_until_booking_time(club, booking_filter)
return await platform.book(booking_filter, club)
def main() -> None:
LOGGER.info("Starting booking padel court")
def main() -> int | None:
user = config.get_user()
booking_filter = config.get_booking_filter()
club = config.get_club()
LOGGER.info(
f"login={config.USER}, password={config.PASSWORD}, club_id={config.CLUB_ID}"
"Starting booking court of %s for user %s at club %s at %s",
booking_filter.sport_id,
user.login,
club.id,
booking_filter.date,
)
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_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")
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

54
resa_padel/config.py
View file

@ -1,21 +1,59 @@
import json
import logging.config
import os
from pathlib import Path
import pendulum
import yaml
from dotenv import load_dotenv
from resa_padel.models import BookingFilter, Club, User
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 get_club() -> 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(),
)
def get_booking_filter() -> BookingFilter:
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")
date_time = pendulum.parse(date_time_tmp) if date_time_tmp else None
return BookingFilter(sport_id=sport_id, date=date_time)
def get_user() -> User:
login = os.environ.get("LOGIN")
password = os.environ.get("PASSWORD")
return User(login=login, password=password)
def get_post_headers(platform_id: str) -> dict:
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:
headers = json.load(f)
return headers
def init_log_config():

55
resa_padel/gestion_sports/gestion_sports_connector.py
View file

@ -1,25 +1,20 @@
import asyncio
import json
import logging
from urllib.parse import urljoin
from aiohttp import ClientResponse, ClientSession
from gestion_sports.gestion_sports_payload_builder import \
GestionSportsPayloadBuilder
from models import BookingFilter, User
import config
from gestion_sports.gestion_sports_payload_builder import GestionSportsPayloadBuilder
from models import BookingFilter, Club, User
DATE_FORMAT = "%d/%m/%Y"
TIME_FORMAT = "%H:%M"
LOGGER = logging.getLogger(__name__)
HEADERS = {
"Connection": "keep-alive",
"Accept-Language": "en-US,en;q=0.5",
"Accept-Encoding": "gzip, deflate, br",
"DNT": "1",
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"Accept": "application/json, text/javascript, */*; q=0.01",
"X-Requested-With": "XMLHttpRequest",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
}
POST_HEADERS = config.get_post_headers("gestion-sports")
class GestionSportsConnector:
@ -32,15 +27,15 @@ class GestionSportsConnector:
@property
def connection_url(self) -> str:
return f"{self.url}/connexion.php?"
return urljoin(self.url, "/connexion.php?")
@property
def login_url(self) -> str:
return f"{self.url}/connexion.php?"
return urljoin(self.url, "/connexion.php?")
@property
def booking_url(self) -> str:
return f"{self.url}/membre/reservation.html?"
return urljoin(self.url, "/membre/reservation.html?")
async def connect(self) -> ClientResponse:
LOGGER.info("Connecting to GestionSports API")
@ -48,29 +43,35 @@ class GestionSportsConnector:
await response.text()
return response
async def login(self, user: User) -> ClientResponse:
async def login(self, user: User, club: Club) -> ClientResponse:
payload = (
self.payload_builder.login(user.login)
.password(user.password)
.club_id(user.club_id)
.club_id(club.id)
.build_login_payload()
)
async with self.session.post(
self.login_url, data=payload, headers=HEADERS
self.login_url, data=payload, headers=POST_HEADERS
) as response:
await response.text()
return response
async def book(self, booking_filter: BookingFilter) -> int | None:
async def book(self, booking_filter: BookingFilter, club: Club) -> int | None:
# 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 booking_filter.court_ids
for court_id in club.courts_ids
],
return_exceptions=True,
)
return await self.get_booked_court(bookings)
@staticmethod
async def get_booked_court(bookings):
for court, is_booked in bookings:
if is_booked:
return court
@ -79,11 +80,9 @@ class GestionSportsConnector:
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))
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(court_id)
.build_booking_payload()
@ -92,7 +91,7 @@ class GestionSportsConnector:
async def is_court_booked(self, payload: str) -> bool:
async with self.session.post(
self.booking_url, data=payload, headers=HEADERS
self.booking_url, data=payload, headers=POST_HEADERS
) as response:
return self.is_response_status_ok(await response.text())

26
resa_padel/models.py
View file

@ -1,16 +1,22 @@
from typing import List
from pydantic import BaseModel, Field
from pendulum import Time
from pydantic import BaseModel, Field, ConfigDict
from pydantic_extra_types.pendulum_dt import DateTime
class Club(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))
class BookingFilter(BaseModel):
sport_id: int = Field()
date: DateTime = Field()
class User(BaseModel):
login: str = Field()
password: str = Field(repr=False)
club_id: str = Field()
class BookingFilter(BaseModel):
court_ids: List[int] = Field()
sport_id: int = Field()
date: DateTime = Field()

12
resa_padel/resources/gestion-sports/post-headers.json Normal file
View file

@ -0,0 +1,12 @@
{
"Connection": "keep-alive",
"Accept-Language": "en-US,en;q=0.5",
"Accept-Encoding": "gzip, deflate, br",
"DNT": "1",
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"Accept": "application/json, text/javascript, */*; q=0.01",
"X-Requested-With": "XMLHttpRequest",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin"
}

31
tests/fixtures.py
View file

@ -2,20 +2,21 @@ 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
from resa_padel.gestion_sports.gestion_sports_payload_builder import (
GestionSportsPayloadBuilder,
)
from resa_padel.models import BookingFilter, Club, 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
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)
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_failure_response = json.dumps(
{
@ -45,10 +46,6 @@ booking_payload = (
.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:
@ -60,6 +57,11 @@ 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
@ -73,8 +75,3 @@ def a_booking_failure_response() -> str:
@pytest.fixture
def a_booking_payload() -> str:
return booking_payload
@pytest.fixture
def a_connector():
return gs_connector

88
tests/gestion_sports/test_gestion_sports_connector.py
View file

@ -2,60 +2,64 @@ import pytest
from aiohttp import ClientSession
from yarl import URL
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)
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_club,
a_user,
)
gestion_sports_url = "https://toulousepadelclub.gestion-sports.com"
tpc_url = "https://toulousepadelclub.gestion-sports.com"
@pytest.mark.asyncio
async def test_should_connect_to_gestion_sports_website():
async with ClientSession() as session:
cookies = session.cookie_jar.filter_cookies(URL(gestion_sports_url))
cookies = session.cookie_jar.filter_cookies(URL(tpc_url))
assert cookies.get("PHPSESSID") is None
gs_connector = GestionSportsConnector(session, gestion_sports_url)
gs_connector = GestionSportsConnector(session, tpc_url)
response = await gs_connector.connect()
assert response.status == 200
assert response.request_info.method == "GET"
assert response.content_type == "text/html"
assert response.request_info.url == URL(gestion_sports_url + "/connexion.php")
assert response.request_info.url == URL(tpc_url + "/connexion.php")
assert response.charset == "UTF-8"
cookies = session.cookie_jar.filter_cookies(URL(gestion_sports_url))
cookies = session.cookie_jar.filter_cookies(URL(tpc_url))
assert cookies.get("PHPSESSID") is not None
@pytest.mark.asyncio
async def test_should_login_to_gestion_sports_website(a_user):
async def test_should_login_to_gestion_sports_website(a_user, a_club):
async with ClientSession() as session:
gs_connector = GestionSportsConnector(session, gestion_sports_url)
gs_connector = GestionSportsConnector(session, tpc_url)
await gs_connector.connect()
response = await gs_connector.login(a_user)
response = await gs_connector.login(a_user, a_club)
assert response.status == 200
assert response.request_info.url == URL(gestion_sports_url + "/connexion.php")
assert response.request_info.url == URL(tpc_url + "/connexion.php")
assert response.request_info.method == "POST"
cookies = session.cookie_jar.filter_cookies(URL(gestion_sports_url))
cookies = session.cookie_jar.filter_cookies(URL(tpc_url))
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_booking_url_should_be_reachable(a_user, a_booking_filter):
async def test_booking_url_should_be_reachable(a_user, a_booking_filter, a_club):
async with ClientSession() as session:
gs_connector = GestionSportsConnector(session, gestion_sports_url)
gs_connector = GestionSportsConnector(session, tpc_url)
await gs_connector.connect()
await gs_connector.login(a_user)
await gs_connector.login(a_user, a_club)
court_booked = await gs_connector.book(a_booking_filter)
court_booked = await gs_connector.book(a_booking_filter, a_club)
# At 18:00 no chance to get a booking, any day of the week
assert court_booked is None
@ -64,24 +68,25 @@ async def test_booking_url_should_be_reachable(a_user, a_booking_filter):
async def test_should_book_a_court_from_gestion_sports(
aioresponses,
a_booking_filter,
a_club,
a_booking_success_response,
a_booking_failure_response,
):
booking_url = URL(gestion_sports_url + "/membre/reservation.html?")
booking_url = URL(tpc_url + "/membre/reservation.html?")
# first booking request will fail
aioresponses.post(URL(booking_url), status=200, body=a_booking_failure_response)
aioresponses.post(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)
aioresponses.post(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)
aioresponses.post(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)
gs_connector = GestionSportsConnector(session, tpc_url)
court_booked = await gs_connector.book(a_booking_filter, a_club)
# the second element of the list is the booked court
assert court_booked == a_booking_filter.court_ids[1]
assert court_booked == a_club.courts_ids[1]
def test_response_status_should_be_ok(a_booking_success_response):
@ -89,28 +94,33 @@ def test_response_status_should_be_ok(a_booking_success_response):
assert is_booked
def test_response_status_should_be_not_ok(aioresponses, a_booking_failure_response):
def test_response_status_should_be_not_ok(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, 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
async with ClientSession() as session:
tpc_connector = GestionSportsConnector(session, tpc_url)
aioresponses.post(
URL(tpc_connector.booking_url), status=200, body=a_booking_failure_response
)
is_booked = await tpc_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, 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
async with ClientSession() as session:
tpc_connector = GestionSportsConnector(session, tpc_url)
aioresponses.post(
URL(tpc_connector.booking_url), status=200, body=a_booking_success_response
)
is_booked = await tpc_connector.is_court_booked(a_booking_payload)
assert is_booked

5
tests/gestion_sports/test_gestion_sports_payload_builder.py
View file

@ -1,7 +1,8 @@
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():

259
tests/test_booking.py
View file

@ -1,66 +1,229 @@
import asyncio
import os
from unittest.mock import patch
from urllib.parse import urljoin
import pendulum
from aioresponses import aioresponses
from pendulum import DateTime, Time
from models import BookingFilter, Club, User
from resa_padel import booking
from resa_padel.models import BookingFilter, User
from tests.fixtures import (a_booking_failure_response,
a_booking_success_response)
from tests import fixtures
from tests.fixtures import (
a_booking_failure_response,
a_booking_filter,
a_booking_success_response,
a_club,
a_user,
)
login = "user"
password = "password"
club_id = "98"
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)
)
# FIXME
# 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(
a_booking_success_response, a_booking_failure_response
def mock_successful_connection(aio_mock, url):
"""
Mock a call to the connection endpoint
:param aio_mock: the aioresponses mock object
:param url: the URL of the connection endpoint
"""
aio_mock.get(
url,
status=200,
headers={"Set-Cookie": f"connection_called=True; Domain={url}"},
)
def mock_successful_login(aio_mock, url):
"""
Mock a call to the login endpoint
:param aio_mock: the aioresponses mock object
:param url: the URL of the login endpoint
"""
aio_mock.post(
url,
status=200,
headers={"Set-Cookie": f"login_called=True; Domain={url}"},
)
def mock_booking(aio_mock, url, response):
"""
Mock a call to the booking endpoint
:param aio_mock: the aioresponses mock object
:param url: the URL of the booking endpoint
:param response: the response from the booking endpoint
"""
aio_mock.post(
url,
status=200,
headers={"Set-Cookie": f"booking_called=True; Domain={url}"},
body=response,
)
def mock_rest_api_from_connection_to_booking(
aio_mock, url: str, a_booking_failure_response: str, a_booking_success_response: str
):
# mock connection to the booking platform
platform_url = "https://some.url"
connection_url = platform_url + "/connexion.php"
login_url = connection_url
booking_url = platform_url + "/membre/reservation.html"
"""
Mock a REST API from a club.
It mocks the calls to the connexion to the website, a call to log in the user
and 2 calls to the booking endpoint
loop = asyncio.get_event_loop()
:param mock_now: the pendulum.now() mock
:param url: the API root URL
:param a_booking_success_response: the success json response
:param a_booking_failure_response: the failure json response
:return:
"""
connexion_url = urljoin(url, "/connexion.php?")
mock_successful_connection(aio_mock, connexion_url)
login_url = urljoin(url, "/connexion.php?")
mock_successful_login(aio_mock, login_url)
booking_url = urljoin(url, "/membre/reservation.html?")
mock_booking(aio_mock, booking_url, a_booking_failure_response)
mock_booking(aio_mock, booking_url, a_booking_success_response)
@patch("pendulum.now")
def test_wait_until_booking_time(
mock_now, a_club: Club, a_booking_filter: BookingFilter
):
"""
Test the function that waits until the booking can be performed
:param mock_now: the pendulum.now() mock
:param a_club: a club
:param a_booking_filter: a booking filter
"""
booking_datetime = retrieve_booking_datetime(a_booking_filter, a_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
booking.wait_until_booking_time(a_club, a_booking_filter)
assert pendulum.now() == booking_datetime.add(microseconds=1)
def retrieve_booking_datetime(
a_booking_filter: BookingFilter, a_club: Club
) -> DateTime:
"""
Utility to retrieve the booking datetime from the booking filter and the club
: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
date_to_book = a_booking_filter.date
return date_to_book.subtract(days=a_club.booking_open_days_before).at(
booking_hour, booking_minute
)
@patch("pendulum.now")
def test_booking_does_the_rights_calls(
mock_now,
a_booking_success_response: str,
a_booking_failure_response: str,
a_user: User,
a_club: Club,
a_booking_filter: BookingFilter,
):
"""
Test a single court booking without reading the conf from environment variables
: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
:param a_user: a test user
:param a_club:a test club
:param a_booking_filter: a test booking filter
"""
booking_datetime = retrieve_booking_datetime(a_booking_filter, a_club)
mock_now.side_effect = [booking_datetime]
# mock connection to the booking platform
with aioresponses() as aio_mock:
mock_rest_api_from_connection_to_booking(
aio_mock,
fixtures.url,
a_booking_failure_response,
a_booking_success_response,
)
loop = asyncio.get_event_loop()
court_booked = loop.run_until_complete(
booking.book(a_club, a_user, a_booking_filter)
)
assert court_booked == a_club.courts_ids[1]
@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(),
},
clear=True,
)
def test_main(
mock_now, a_booking_success_response: str, a_booking_failure_response: str
):
"""
Test the main function to book a court
: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 = retrieve_booking_datetime(booking_filter, club)
mock_now.side_effect = [booking_datetime]
with aioresponses() as aio_mock:
aio_mock.get(
connection_url,
status=200,
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={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,
mock_rest_api_from_connection_to_booking(
aio_mock,
fixtures.url,
a_booking_failure_response,
a_booking_success_response,
)
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)
)
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"
court_booked = booking.main()
assert court_booked == 8