Skip to content

Commit

Permalink
(PC-31471) test: add assert_num_queries in public
Browse files Browse the repository at this point in the history
  • Loading branch information
@blbpro @R0ntheo
blbpro authored and R0ntheo committed Sep 3, 2024
1 parent aa9e2d9 commit d84c4a4
Show file tree
Hide file tree
Showing 27 changed files with 1,062 additions and 638 deletions.
6 changes: 5 additions & 1 deletion api/src/pcapi/core/educational/repository.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1079,7 +1079,11 @@ def get_educational_domains_from_ids(ids: typing.Iterable[int]) -> list[educatio


def get_all_educational_domains_ordered_by_name() -> list[educational_models.EducationalDomain]:
return educational_models.EducationalDomain.query.order_by(educational_models.EducationalDomain.name).all()
return (
educational_models.EducationalDomain.query.order_by(educational_models.EducationalDomain.name)
.options(sa.orm.joinedload(educational_models.EducationalDomain.nationalPrograms))
.all()
)


def get_all_educational_institutions(offset: int = 0, limit: int = 0) -> tuple[tuple, int]:
Expand Down
7 changes: 6 additions & 1 deletion api/src/pcapi/core/providers/repository.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -194,7 +194,12 @@ def is_cinema_external_ticket_applicable(offer: offers_models.Offer) -> bool:


def get_providers_venues(provider_id: int) -> BaseQuery:
return Venue.query.join(models.VenueProvider).filter(models.VenueProvider.providerId == provider_id)
return (
Venue.query.join(models.VenueProvider)
.outerjoin(models.VenueProvider.externalUrls)
.options(sqla_orm.contains_eager(Venue.venueProviders).contains_eager(models.VenueProvider.externalUrls))
.filter(models.VenueProvider.providerId == provider_id)
)


def _get_future_provider_events_requiring_a_ticketing_system_query(
Expand Down
179 changes: 111 additions & 68 deletions api/tests/routes/public/booking_token/v2/get_booking_by_token_v2_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,21 +4,29 @@
from dateutil.relativedelta import relativedelta
import pytest

from pcapi.core import testing
import pcapi.core.bookings.factories as bookings_factories
from pcapi.core.categories import subcategories_v2 as subcategories
import pcapi.core.offerers.factories as offerers_factories
import pcapi.core.offers.factories as offers_factories
from pcapi.utils.date import format_into_utc_date
from pcapi.utils.date import utc_datetime_to_department_timezone
from pcapi.utils import date
from pcapi.utils.human_ids import humanize


pytestmark = pytest.mark.usefixtures("db_session")


class Returns200Test:
num_queries = 1 # select user
num_queries += 1 # select booking
num_queries += 1 # check user has rights on offerer
num_queries += 1 # check if a pricing processed or invoiced exists for this booking
num_queries += 1 # select stock
num_queries += 1 # select offer
num_queries += 1 # select user
num_queries += 1 # select venue

def test_when_user_has_rights_and_regular_offer(self, client):
# Given
past = datetime.utcnow() - timedelta(days=2)
booking = bookings_factories.BookingFactory(
user__email="[email protected]",
Expand All @@ -40,17 +48,17 @@ def test_when_user_has_rights_and_regular_offer(self, client):
user_offerer = offerers_factories.UserOffererFactory(offerer=booking.offerer)
pro_user = user_offerer.user

# When
client = client.with_basic_auth(pro_user.email)
response = client.get(f"/v2/bookings/token/{booking.token}")
booking_token = booking.token
with testing.assert_num_queries(self.num_queries):
response = client.get(f"/v2/bookings/token/{booking_token}")
assert response.status_code == 200

# Then
assert response.headers["Content-type"] == "application/json"
assert response.status_code == 200
assert response.json == {
"bookingId": humanize(booking.id),
"dateOfBirth": booking.user.birth_date.isoformat(),
"datetime": format_into_utc_date(booking.stock.beginningDatetime),
"datetime": date.format_into_utc_date(booking.stock.beginningDatetime),
"ean13": None,
"email": "[email protected]",
"formula": "ABO",
Expand All @@ -76,110 +84,124 @@ def test_when_user_has_rights_and_regular_offer(self, client):
}

def test_when_api_key_is_provided_and_rights_and_regular_offer(self, client):
# Given
booking = bookings_factories.BookingFactory()
booking_token = booking.token
offerers_factories.ApiKeyFactory(offerer=booking.offerer, prefix="test_prefix")

# When
url = f"/v2/bookings/token/{booking.token}"
response = client.get(url, headers={"Authorization": "Bearer test_prefix_clearSecret"})

# Then
assert response.status_code == 200
url = f"/v2/bookings/token/{booking_token}"
with testing.assert_num_queries(self.num_queries):
response = client.get(url, headers={"Authorization": "Bearer test_prefix_clearSecret"})
assert response.status_code == 200

def test_when_user_has_rights_and_regular_offer_and_token_in_lower_case(self, client):
# Given
booking = bookings_factories.BookingFactory(
stock__beginningDatetime=datetime.utcnow() - timedelta(days=2),
)
booking_token = booking.token.lower()
user_offerer = offerers_factories.UserOffererFactory(offerer=booking.offerer)
pro_user = user_offerer.user

# When
client = client.with_basic_auth(pro_user.email)
response = client.get(f"/v2/bookings/token/{booking.token.lower()}")

# Then
assert response.status_code == 200
with testing.assert_num_queries(self.num_queries):
response = client.get(f"/v2/bookings/token/{booking_token}")
assert response.status_code == 200


class NonStandardGetTest:
num_queries = 1 # select user
num_queries += 1 # select booking
num_queries += 1 # check user has rights on offerer
num_queries += 1 # check if a pricing processed or invoiced exists for this booking
num_queries += 1 # select stock
num_queries += 1 # select offer
num_queries += 1 # select user
num_queries += 1 # select venue

def test_non_standard_get_on_token_endpoint(self, client):
# This is a test following the incident caused by a check on the JSON sent by API user (PR #12928 introduced the bug, PR #13062 fixed it)
# Some legacy users are sending us an invalid JSON in a GET request to /v2/bookings/token/<token>
# We must not raise an error in those cases otherwise it breaks their integrations.
booking = bookings_factories.BookingFactory()
offerers_factories.ApiKeyFactory(offerer=booking.offerer, prefix="test_prefix")

# When
url = f"/v2/bookings/token/{booking.token}"
response = client.get_with_invalid_json_body(
url,
headers={"Authorization": "Bearer test_prefix_clearSecret"},
raw_json="ABC", # both sad an irritating
)

# Then
assert response.status_code == 200
with testing.assert_num_queries(self.num_queries):
response = client.get_with_invalid_json_body(
url,
headers={"Authorization": "Bearer test_prefix_clearSecret"},
raw_json="ABC", # both sad an irritating
)
assert response.status_code == 200


class Returns401Test:
def test_when_user_no_auth_nor_api_key(self, client):
response = client.get("/v2/bookings/token/TOKEN")
assert response.status_code == 401
with testing.assert_num_queries(0):
response = client.get("/v2/bookings/token/TOKEN")
assert response.status_code == 401

def test_when_wrong_api_key(self, client):
url = "/v2/bookings/token/FAKETOKEN"
response = client.get(url, headers={"Authorization": "Bearer WrongApiKey1234567"})
assert response.status_code == 401
num_queries = 1 # select api_key
with testing.assert_num_queries(num_queries):
response = client.get(url, headers={"Authorization": "Bearer WrongApiKey1234567"})
assert response.status_code == 401


class Returns403Test:
def test_when_user_doesnt_have_rights_and_token_exists(self, client):
# Given
booking = bookings_factories.BookingFactory()
another_pro_user = offerers_factories.UserOffererFactory().user

# When
url = f"/v2/bookings/token/{booking.token}"
response = client.with_basic_auth(another_pro_user.email).get(url)
client = client.with_basic_auth(another_pro_user.email)
num_queries = 1 # select user
num_queries += 1 # select booking
num_queries += 1 # check user has rights on offerer
with testing.assert_num_queries(num_queries):
response = client.get(url)
assert response.status_code == 403

# Then
assert response.status_code == 403
assert response.json["user"] == [
"Vous n’avez pas les droits suffisants pour valider cette contremarque car cette réservation n'a pas été faite sur une de vos offres, ou que votre rattachement à la structure est encore en cours de validation"
]

def test_when_given_api_key_not_related_to_booking_offerer(self, client):
# Given
booking = bookings_factories.BookingFactory()
offerers_factories.ApiKeyFactory() # another offerer's API key

# When
auth = "Bearer development_prefix_clearSecret"
url = f"/v2/bookings/token/{booking.token}"
response = client.get(url, headers={"Authorization": auth})
num_queries = 1 # select api_key
num_queries += 1 # select feature
num_queries += 1 # select booking
with testing.assert_num_queries(num_queries):
response = client.get(url, headers={"Authorization": auth})
assert response.status_code == 403

# Then
assert response.status_code == 403
assert response.json["user"] == [
"Vous n’avez pas les droits suffisants pour valider cette contremarque car cette réservation n'a pas été faite sur une de vos offres, ou que votre rattachement à la structure est encore en cours de validation"
]

def test_when_booking_not_confirmed(self, client):
# Given
next_week = datetime.utcnow() + timedelta(weeks=1)
booking = bookings_factories.BookingFactory(stock__beginningDatetime=next_week)
pro_user = offerers_factories.UserOffererFactory(offerer=booking.offerer).user

# When
url = f"/v2/bookings/token/{booking.token}"
response = client.with_basic_auth(pro_user.email).get(url)
num_queries = 1 # Select user
num_queries += 1 # Select booking
num_queries += 1 # check user has rights on offerer
num_queries += 1 # check if a pricing processed or invoiced exists for this booking
num_queries += 1 # select stock
num_queries += 1 # select venue
client = client.with_basic_auth(pro_user.email)
with testing.assert_num_queries(num_queries):
response = client.get(url)
assert response.status_code == 403

# Then
assert response.status_code == 403
cancellation_limit_date = datetime.strftime(
utc_datetime_to_department_timezone(booking.cancellationLimitDate, booking.venue.departementCode),
date.utc_datetime_to_department_timezone(booking.cancellationLimitDate, booking.venue.departementCode),
"%d/%m/%Y à %H:%M",
)
assert (
Expand All @@ -188,37 +210,49 @@ def test_when_booking_not_confirmed(self, client):
)

def test_when_booking_is_refunded(self, client):
# Given
booking = bookings_factories.ReimbursedBookingFactory()
pro_user = offerers_factories.UserOffererFactory(offerer=booking.offerer).user

# When
url = f"/v2/bookings/token/{booking.token}"
response = client.with_basic_auth(pro_user.email).get(url)
num_queries = 1 # Select user
num_queries += 1 # Select booking
num_queries += 1 # check user has rights on offerer
client = client.with_basic_auth(pro_user.email)
with testing.assert_num_queries(num_queries):
response = client.get(url)
assert response.status_code == 403

# Then
assert response.status_code == 403
assert response.json["payment"] == ["Cette réservation a été remboursée"]


class Returns404Test:
def test_missing_token(self, client):
response = client.get("/v2/bookings/token/")
assert response.status_code == 404
with testing.assert_num_queries(0):
response = client.get("/v2/bookings/token/")
assert response.status_code == 404

def test_basic_auth_but_unknown_token(self, client):
user = offerers_factories.UserOffererFactory().user
client = client.with_basic_auth(user.email)
response = client.get("/v2/bookings/token/UNKNOWN")
num_queries = 1 # select user
num_queries += 1 # select booking
with testing.assert_num_queries(num_queries):
response = client.get("/v2/bookings/token/UNKNOWN")
assert response.status_code == 404

assert response.status_code == 404
assert response.json["global"] == ["Cette contremarque n'a pas été trouvée"]

def test_authenticated_with_api_key_but_token_not_found(self, client):
offerers_factories.ApiKeyFactory(prefix="test_prefix")
response = client.get("/v2/bookings/token/12345", headers={"Authorization": "Bearer test_prefix_clearSecret"})
num_queries = 1 # select api_key
num_queries += 1 # select feature
num_queries += 1 # select booking
with testing.assert_num_queries(num_queries):
response = client.get(
"/v2/bookings/token/12345", headers={"Authorization": "Bearer test_prefix_clearSecret"}
)
assert response.status_code == 404

assert response.status_code == 404
assert response.json["global"] == ["Cette contremarque n'a pas été trouvée"]


Expand All @@ -228,20 +262,29 @@ def test_when_booking_is_already_validated(self, client):
user = offerers_factories.UserOffererFactory(offerer=booking.offerer).user

client = client.with_basic_auth(user.email)
response = client.get(f"/v2/bookings/token/{booking.token}")
num_queries = 1 # Select user
num_queries += 1 # Select booking
num_queries += 1 # check user has rights on offerer
num_queries += 1 # check if a pricing processed or invoiced exists for this booking
token = booking.token
with testing.assert_num_queries(num_queries):
response = client.get(f"/v2/bookings/token/{token}")
assert response.status_code == 410

assert response.status_code == 410
assert response.json["booking"] == ["Cette réservation a déjà été validée"]

def test_when_booking_is_cancelled(self, client):
# Given
booking = bookings_factories.CancelledBookingFactory()
pro_user = offerers_factories.UserOffererFactory(offerer=booking.offerer).user

# When
url = f"/v2/bookings/token/{booking.token}"
response = client.with_basic_auth(pro_user.email).get(url)
num_queries = 1 # Select user
num_queries += 1 # Select booking
num_queries += 1 # check user has rights on offerer
num_queries += 1 # check if a pricing processed or invoiced exists for this booking
client = client.with_basic_auth(pro_user.email)
with testing.assert_num_queries(num_queries):
response = client.get(url)
assert response.status_code == 410

# Then
assert response.status_code == 410
assert response.json["booking_cancelled"] == ["Cette réservation a été annulée"]
Loading

0 comments on commit d84c4a4

Please sign in to comment.