refactored code to place the booking specific actions inside a gestion-sports element
This commit is contained in:
parent
d0a0072b6d
commit
26670bfe34
5 changed files with 251 additions and 215 deletions
|
|
@ -1,53 +1,13 @@
|
|||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
|
||||
import config
|
||||
import pendulum
|
||||
from aiohttp import ClientSession
|
||||
from gestion_sports.gestion_sports_connector import GestionSportsConnector
|
||||
from gestion_sports.gestion_sports_operations import GestionSportsOperations
|
||||
from models import BookingFilter, Club, User
|
||||
from pendulum import DateTime
|
||||
|
||||
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()
|
||||
while now < booking_datetime:
|
||||
time.sleep(1)
|
||||
now = pendulum.now()
|
||||
|
||||
|
||||
def build_booking_datetime(booking_filter: BookingFilter, club: Club) -> DateTime:
|
||||
"""
|
||||
Build the date and time when the booking is open for a given match date.
|
||||
The booking filter contains the date and time of the booking.
|
||||
The club has the information about when the booking is open for that date.
|
||||
|
||||
:param booking_filter: the booking information
|
||||
:param club: the club where to book a court
|
||||
:return: the date and time when the booking is open
|
||||
"""
|
||||
date_to_book = booking_filter.date
|
||||
booking_date = date_to_book.subtract(days=club.booking_open_days_before)
|
||||
|
||||
booking_hour = club.booking_opening_time.hour
|
||||
booking_minute = club.booking_opening_time.minute
|
||||
|
||||
return booking_date.at(booking_hour, booking_minute)
|
||||
|
||||
|
||||
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
|
||||
|
|
@ -57,12 +17,8 @@ async def book(club: Club, user: User, booking_filter: BookingFilter) -> int | N
|
|||
: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.land()
|
||||
await platform.login(user, club)
|
||||
wait_until_booking_time(club, booking_filter)
|
||||
return await platform.book(booking_filter, club)
|
||||
async with GestionSportsOperations(club) as platform:
|
||||
return await platform.book(user, booking_filter)
|
||||
|
||||
|
||||
def main() -> int | None:
|
||||
|
|
|
|||
70
resa_padel/gestion_sports/gestion_sports_operations.py
Normal file
70
resa_padel/gestion_sports/gestion_sports_operations.py
Normal file
|
|
@ -0,0 +1,70 @@
|
|||
import logging
|
||||
import time
|
||||
|
||||
import pendulum
|
||||
from aiohttp import ClientSession
|
||||
from gestion_sports.gestion_sports_connector import GestionSportsConnector
|
||||
from models import BookingFilter, Club, User
|
||||
from pendulum import DateTime
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class GestionSportsOperations:
|
||||
def __init__(self, club: Club):
|
||||
self.platform: GestionSportsConnector = None
|
||||
self.club: Club = club
|
||||
self.session: ClientSession | None = None
|
||||
|
||||
async def __aenter__(self):
|
||||
self.session = ClientSession()
|
||||
self.platform = GestionSportsConnector(self.session, self.club.url)
|
||||
return self
|
||||
|
||||
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
||||
await self.session.close()
|
||||
|
||||
async def book(self, user: User, booking_filter: BookingFilter) -> int | None:
|
||||
if self.platform is None or user is None or booking_filter is None:
|
||||
return None
|
||||
|
||||
await self.platform.land()
|
||||
await self.platform.login(user, self.club)
|
||||
self.wait_until_booking_time(self.club, booking_filter)
|
||||
return await self.platform.book(booking_filter, self.club)
|
||||
|
||||
@staticmethod
|
||||
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()
|
||||
while now < booking_datetime:
|
||||
time.sleep(1)
|
||||
now = pendulum.now()
|
||||
|
||||
|
||||
def build_booking_datetime(booking_filter: BookingFilter, club: Club) -> DateTime:
|
||||
"""
|
||||
Build the date and time when the booking is open for a given match date.
|
||||
The booking filter contains the date and time of the booking.
|
||||
The club has the information about when the booking is open for that date.
|
||||
|
||||
:param booking_filter: the booking information
|
||||
:param club: the club where to book a court
|
||||
:return: the date and time when the booking is open
|
||||
"""
|
||||
date_to_book = booking_filter.date
|
||||
booking_date = date_to_book.subtract(days=club.booking_open_days_before)
|
||||
|
||||
booking_hour = club.booking_opening_time.hour
|
||||
booking_minute = club.booking_opening_time.minute
|
||||
|
||||
return booking_date.at(booking_hour, booking_minute)
|
||||
81
tests/gestion_sports/test_gestion_sports_operations.py
Normal file
81
tests/gestion_sports/test_gestion_sports_operations.py
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
from unittest.mock import patch
|
||||
|
||||
import pendulum
|
||||
import pytest
|
||||
from aioresponses import aioresponses
|
||||
from gestion_sports.gestion_sports_operations import GestionSportsOperations
|
||||
from models import BookingFilter, Club, User
|
||||
|
||||
from tests import fixtures, utils
|
||||
from tests.fixtures import (
|
||||
a_booking_failure_response,
|
||||
a_booking_filter,
|
||||
a_booking_success_response,
|
||||
a_club,
|
||||
a_user,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("pendulum.now")
|
||||
async def test_booking(
|
||||
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 = utils.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:
|
||||
utils.mock_rest_api_from_connection_to_booking(
|
||||
aio_mock,
|
||||
fixtures.url,
|
||||
a_booking_failure_response,
|
||||
a_booking_success_response,
|
||||
)
|
||||
|
||||
async with GestionSportsOperations(a_club) as gs_operations:
|
||||
court_booked = await gs_operations.book(a_user, a_booking_filter)
|
||||
assert court_booked == a_club.courts_ids[1]
|
||||
|
||||
|
||||
@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 = utils.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
|
||||
|
||||
GestionSportsOperations.wait_until_booking_time(a_club, a_booking_filter)
|
||||
|
||||
assert pendulum.now() == booking_datetime.add(microseconds=1)
|
||||
|
|
@ -1,22 +1,14 @@
|
|||
import asyncio
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
from urllib.parse import urljoin
|
||||
|
||||
import pendulum
|
||||
from aioresponses import aioresponses
|
||||
from models import BookingFilter, Club, User
|
||||
from pendulum import DateTime, Time
|
||||
from models import BookingFilter, Club
|
||||
from pendulum import Time
|
||||
|
||||
from resa_padel import booking
|
||||
from tests import fixtures
|
||||
from tests.fixtures import (
|
||||
a_booking_failure_response,
|
||||
a_booking_filter,
|
||||
a_booking_success_response,
|
||||
a_club,
|
||||
a_user,
|
||||
)
|
||||
from tests import fixtures, utils
|
||||
from tests.fixtures import a_booking_failure_response, a_booking_success_response
|
||||
|
||||
login = "user"
|
||||
password = "password"
|
||||
|
|
@ -28,160 +20,6 @@ datetime_to_book = (
|
|||
)
|
||||
|
||||
|
||||
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 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
|
||||
|
||||
: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,
|
||||
|
|
@ -214,11 +52,11 @@ def test_main(
|
|||
booking_open_days_before=7,
|
||||
booking_opening_time=Time(hour=0, minute=0),
|
||||
)
|
||||
booking_datetime = retrieve_booking_datetime(booking_filter, club)
|
||||
booking_datetime = utils.retrieve_booking_datetime(booking_filter, club)
|
||||
mock_now.side_effect = [booking_datetime]
|
||||
|
||||
with aioresponses() as aio_mock:
|
||||
mock_rest_api_from_connection_to_booking(
|
||||
utils.mock_rest_api_from_connection_to_booking(
|
||||
aio_mock,
|
||||
fixtures.url,
|
||||
a_booking_failure_response,
|
||||
|
|
|
|||
91
tests/utils.py
Normal file
91
tests/utils.py
Normal file
|
|
@ -0,0 +1,91 @@
|
|||
from urllib.parse import urljoin
|
||||
|
||||
from models import BookingFilter, Club
|
||||
from pendulum import DateTime
|
||||
|
||||
|
||||
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 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
|
||||
)
|
||||
|
||||
|
||||
def mock_rest_api_from_connection_to_booking(
|
||||
aio_mock, url: str, a_booking_failure_response: str, a_booking_success_response: str
|
||||
):
|
||||
"""
|
||||
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
|
||||
|
||||
:param aio_mock: 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)
|
||||
Loading…
Add table
Add a link
Reference in a new issue