229 lines
6.5 KiB
Python
229 lines
6.5 KiB
Python
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 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 = "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)
|
|
)
|
|
|
|
|
|
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,
|
|
{
|
|
"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:
|
|
mock_rest_api_from_connection_to_booking(
|
|
aio_mock,
|
|
fixtures.url,
|
|
a_booking_failure_response,
|
|
a_booking_success_response,
|
|
)
|
|
|
|
court_booked = booking.main()
|
|
assert court_booked == 8
|