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

Refactoring for reading the config

Browse source
This commit is contained in:
Stanislas Jouffroy 2024-02-17 09:35:49 +01:00
parent 963ee6b86f
commit fc11a1e1eb
9 changed files with 264 additions and 166 deletions
Download patch file Download diff file Expand all files Collapse all files

28
tests/fixtures.py
View file

@ -2,20 +2,20 @@ 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)
booking_filter = BookingFilter(sport_id=sport_id, date=booking_date)
booking_failure_response = json.dumps(
{
@ -45,10 +45,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 +56,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 +74,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():

137
tests/test_booking.py
View file

@ -1,66 +1,109 @@
import asyncio
import os
from unittest.mock import patch
from urllib.parse import urljoin
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)
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"
# 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 mock_successful_connection(aio_mock, url):
aio_mock.get(
url,
status=200,
headers={"Set-Cookie": f"connection_called=True; Domain={url}"},
)
def mock_successful_login(aio_mock, url):
aio_mock.post(
url,
status=200,
headers={"Set-Cookie": f"login_called=True; Domain={url}"},
)
def mock_booking(aio_mock, url, response):
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, a_booking_failure_response, a_booking_success_response
):
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)
def test_booking_does_the_rights_calls(
a_booking_success_response, a_booking_failure_response
a_booking_success_response,
a_booking_failure_response,
a_user,
a_club,
a_booking_filter,
):
# 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"
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()
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.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": "2024-04-23T15:00:00Z",
},
clear=True,
)
def test_main(a_booking_success_response, a_booking_failure_response):
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