From 5c57c9706e30f6f7d18a59eef5012f453f778310 Mon Sep 17 00:00:00 2001 From: Lukas Bockstaller Date: Mon, 20 Jul 2026 16:51:56 +0200 Subject: [PATCH] tests --- src/pretix/base/models/cancellation.py | 114 +++++--- src/pretix/base/services/orders.py | 125 +++----- src/pretix/helpers/database.py | 8 +- src/pretix/plugins/ticketoutputpdf/signals.py | 12 +- .../base/test_self_service_cancellation.py | 266 +++++++++++++++++- src/tests/helpers/test_database.py | 77 +++++ 6 files changed, 453 insertions(+), 149 deletions(-) create mode 100644 src/tests/helpers/test_database.py diff --git a/src/pretix/base/models/cancellation.py b/src/pretix/base/models/cancellation.py index 21fdf57ebd..e265e50a0a 100644 --- a/src/pretix/base/models/cancellation.py +++ b/src/pretix/base/models/cancellation.py @@ -1,8 +1,10 @@ -from abc import ABC from dataclasses import dataclass, field from decimal import Decimal from itertools import chain -from typing import Callable, Dict, List, Literal, Optional, Protocol, Set, TYPE_CHECKING, TypeAlias +from typing import ( + TYPE_CHECKING, Any, Callable, Dict, List, Literal, Optional, Protocol, Set, + Tuple, TypeAlias, +) from django.core.exceptions import ValidationError from django.core.validators import MaxValueValidator, MinValueValidator @@ -18,9 +20,9 @@ from pretix.base.signals import self_service_cancellation_checks """ Supporting self-service cancellation requires us to do two main things: 1. uphold the business logic of pretix and the installed plugins -2. charge the customer the appropriate fees for their cancellation +2. charge the customer the appropriate fees for their cancellation -Number 1 is a question of bringing enough checks into place and prevent a +Number 1 is a question of bringing enough checks into place and prevent a cancellation if one of them is violated. Checks need to subclass `CancellationCheck` and can be provided via the new `self_service_cancellation_checks` signal. @@ -33,26 +35,26 @@ The cancellation fees are computed via `CancellationRules`. When a customer triggers a self service cancellation, we will: 1. Positions - a. Evaluate all `CancellationChecks` that are concerned with individual positions + a. Evaluate all `CancellationChecks` that are concerned with individual positions b. Evaluate all `CancellationRules` that are concerned with individual positions and compute the fees c. Choose for each position the cheapest `CancellationRules` position result available -2. Process +2. Process a. Evaluate all `CancellationChecks` that are concerned with the process of cancellation b. Evaluate all `CancellationRules` that are concerned with the process of cancellation c. Choose the cheapest `CancellationRules` process result available -3. Return all results for Checks and Rules +3. Return all results for Checks and Rules -Step 1c. and 2c. are kept separate intentionally. -The alternative of finding the cheapest cancellation option overall (process and position) would +Step 1c. and 2c. are kept separate intentionally. +The alternative of finding the cheapest cancellation option overall (process and position) would require us to check the full combinatorics of possible process and position fees, resulting -in unfeasable runtime behaviour, and if we would optimize it in difficult to explain non-optimal -situations. +in unfeasible runtime behaviour, and if we would optimize it in difficult to explain non-optimal +situations. """ class FeeType(models.TextChoices): """ - Process fees can be added on top of all position fees or they + Process fees can be added on top of all position fees, or they can set a floor for the minimum cancellation fee that this will incur. """ MINIMUM = "min_process_fee", _("Minimum total fee") @@ -84,7 +86,7 @@ class RuleResult: Result of evaluating a CancellationRule. A rule can consist out of multiple different checks, each partial_result is recorded individually. - A RuleResult encodes both the feasibility of a cancellation via `cancellation_possible` as well as + A RuleResult encodes both the feasibility of a cancellation via `cancellation_possible` and the resulting consequences in form of fees which can be expressed as: - absolute position fees of a fixed amount - relative position fees of a percentage of the position price @@ -139,13 +141,15 @@ class RuleResult: if reference_price < absolute_fee: fee = absolute_fee - reference_price else: - fee = reference_price + fee = Decimal(0) elif fee_type == FeeType.ADDITIONAL: fee = absolute_fee + else: + raise ValueError("Unknown fee type") return RuleResult(id=id, partial_results=partial_results, fee_type=fee_type, fee=fee) - def __lt__(self, other): + def __lt__(self, other: object) -> bool: if not isinstance(other, RuleResult): return NotImplemented @@ -168,20 +172,25 @@ class Checks: def related_selects(self) -> List[str]: return list(chain([check.related_selects for check in [*self.position, *self.process]])) + PositionSet: TypeAlias = Set[OrderPosition] + class PositionCheckFn(Protocol): - def __call__(self, order: Order, keep: PositionSet, position: OrderPosition) -> CheckResult: ... + def __call__(self, order: Order, keep: PositionSet, position: OrderPosition) -> CheckResult: + ... + class ProcessCheckFn(Protocol): - def __call__(self, order: Order, keep: PositionSet) -> CheckResult: ... + def __call__(self, order: Order, keep: PositionSet) -> CheckResult: + ... @dataclass(frozen=True) -class CancellationCheck(ABC): +class CancellationCheck: id: str type: CheckTypes - check_fn: PositionCheckFn | ProcessCheckFn + check_fn: PositionCheckFn | ProcessCheckFn = field(compare=False) prefetches: List[Callable[[], Prefetch]] = field(default_factory=list) related_selects: List[str] = field(default_factory=list) @@ -200,22 +209,24 @@ class PositionResult: position_check_results: Dict[int, List[CheckResult]] position_rule_results: Dict[int, List[RuleResult]] + @property def cancellation_possible(self) -> bool: - def ok(results: list) -> bool: - return results[0].cancellation_possible if results else True + def ok(results: List[CheckResult] | List[RuleResult]) -> bool: + return all([val.cancellation_possible for val in results]) if results else True return all( ok(results) - for d in (self.position_check_results, self.position_rule_results) + for d in + (self.position_check_results, {key: [min(pos_res)] for key, pos_res in self.position_rule_results.items()}) for results in d.values() ) + @property def fee_value(self) -> Decimal: fee_value = Decimal("0.00") for pos_id, results in self.position_rule_results.items(): if len(results) > 0: - results.sort() - best_option = results[0] + best_option = min(results) if best_option.cancellation_possible: fee_value += best_option.fee return fee_value @@ -226,8 +237,17 @@ class ProcessResult: process_check_results: List[CheckResult] process_rule_results: List[RuleResult] + @property def cancellation_possible(self) -> bool: - return all([res.cancellation_possible for res in [*self.process_check_results, *self.process_rule_results]]) + best_option = min(self.process_rule_results) + return all([res.cancellation_possible for res in [*self.process_check_results, best_option]]) + + @property + def fee_value(self) -> Decimal: + best_option = min(self.process_rule_results) + if best_option.cancellation_possible: + return best_option.fee + return Decimal("0.00") @dataclass(frozen=True) @@ -235,8 +255,9 @@ class CancellationResult: position_result: PositionResult process_result: ProcessResult + @property def cancellation_possible(self) -> bool: - return self.position_result.cancellation_possible() and self.process_result.cancellation_possible() + return self.position_result.cancellation_possible and self.process_result.cancellation_possible def remember_cancellation(self): # TODO: store the cancellation verdict in the session storage for X Minutes @@ -247,6 +268,10 @@ class CancellationResult: pass +def _send_self_service_cancellation_checks(event: Event) -> List[Tuple[Any, Any]]: + return self_service_cancellation_checks.send(sender=event) + + class CancellationRule(models.Model): event = models.ForeignKey( Event, @@ -269,13 +294,15 @@ class CancellationRule(models.Model): related_selects: List[str] = [] @staticmethod - def _collect_checks(event: Event) -> Checks: + def _collect_checks(event: Event, send_fn: Callable[ + [Event], List[Tuple[Any, Any]] + ] = _send_self_service_cancellation_checks) -> Checks: position_checks: List[CancellationCheck] = [] process_checks: List[CancellationCheck] = [] seen = set() - for recv, resp in self_service_cancellation_checks.send(sender=event): - if not isinstance(recv, CancellationCheck): + for recv, resp in send_fn(event): + if not isinstance(resp, CancellationCheck): raise ValueError('self_service_cancellation_checks received response of wrong type') if resp.id in seen: raise ValueError('self_service_cancellation_checks received multiple responses with the id') @@ -295,15 +322,7 @@ class CancellationRule(models.Model): position_rules = PositionCancellationRule.objects.filter(event=event, type=CheckTypes.POSITION) process_rules = ProcessCancellationRule.objects.filter(event=event, type=CheckTypes.PROCESS) - # prefetch and join everything these rules want - prefetches = [*checks.prefetches, - *PositionCancellationRule.prefetches, - *ProcessCancellationRule.prefetches] - related_selects = list(set(*checks.related_selects, - *PositionCancellationRule.related_selects, - *ProcessCancellationRule.related_selects)) - order = Order.objects.prefetch_related(*prefetches).select_related(*related_selects).get(event=event, - id=order.id) + order = CancellationRule._prefetch_order(event, order, checks) # keep track of all decisions so we can explain them in the logs position_check_results: Dict[int, List[CheckResult]] = {} @@ -332,7 +351,7 @@ class CancellationRule(models.Model): position_rule_results=position_rule_results) # we need the current fee_value to select the cheapest process rule - temp_position_fees = position_results.fee_value() + temp_position_fees = position_results.fee_value # again keep track of all decisions so we can explain them in the logs process_check_results: List[CheckResult] = [] @@ -353,6 +372,20 @@ class CancellationRule(models.Model): return CancellationResult(position_result=position_results, process_result=process_result) + @staticmethod + def _prefetch_order(event: Event, order: Order, checks: Checks) -> Order: + prefetches = [pref() for pref in [*chain(*checks.prefetches), + *chain(*PositionCancellationRule.prefetches), + *chain(*ProcessCancellationRule.prefetches)]] + + related_selects = {*chain(*checks.related_selects), + *chain(*PositionCancellationRule.related_selects), + *chain(*ProcessCancellationRule.related_selects)} + + order = Order.objects.prefetch_related(*prefetches).select_related(*related_selects).get(event=event, + id=order.id) + return order + def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @@ -398,7 +431,8 @@ class PositionCancellationRule(CancellationRule): except_after = ModelRelativeDateTimeField(null=True, blank=True) def evaluate_position_rule(self, order: Order, keep: Set[OrderPosition], position: OrderPosition) -> Optional[ - RuleResult]: + RuleResult + ]: if not self.all_products and position.item_id not in self.limit_products.values_list('pk', flat=True): return None diff --git a/src/pretix/base/services/orders.py b/src/pretix/base/services/orders.py index 33e049aead..4e4e09829d 100644 --- a/src/pretix/base/services/orders.py +++ b/src/pretix/base/services/orders.py @@ -41,16 +41,19 @@ from collections import Counter, defaultdict, namedtuple from datetime import datetime, time, timedelta from decimal import Decimal from functools import reduce -from itertools import chain + from time import sleep -from typing import Dict, List, Optional, Set +from typing import List, Optional from celery.exceptions import MaxRetriesExceededError from django.conf import settings from django.core.cache import cache from django.core.exceptions import ValidationError from django.db import transaction -from django.db.models import Count, Exists, F, IntegerField, Max, Min, OuterRef, Q, QuerySet, Sum, Value +from django.db.models import ( + Count, Exists, F, IntegerField, Max, Min, OuterRef, Prefetch, Q, QuerySet, + Sum, Value, +) from django.db.models.functions import Coalesce, Greatest from django.db.transaction import get_connection from django.dispatch import receiver @@ -69,14 +72,16 @@ from pretix.base.models import ( Membership, Order, OrderPayment, OrderPosition, Quota, Seat, SeatCategoryMapping, User, Voucher, ) -from pretix.base.models.cancellation import (CancellationCheck, CancellationRule) +from pretix.base.models.cancellation import ( + CancellationCheck, CheckResult, CheckTypes, PositionSet, +) from pretix.base.models.event import SubEvent from pretix.base.models.orders import ( BlockedTicketSecret, InvoiceAddress, OrderFee, OrderRefund, generate_secret, ) from pretix.base.models.organizer import SalesChannel, TeamAPIToken -from pretix.base.models.tax import TAXED_ZERO, TaxRule, TaxedPrice +from pretix.base.models.tax import TAXED_ZERO, TaxedPrice, TaxRule from pretix.base.payment import GiftCardPayment, PaymentException from pretix.base.reldate import RelativeDateWrapper from pretix.base.secrets import assign_ticket_secret @@ -106,7 +111,7 @@ from pretix.base.signals import ( ) from pretix.base.timemachine import time_machine_now, time_machine_now_assigned from pretix.celery_app import app -from pretix.helpers import OF_SELF, ensure_no_queries +from pretix.helpers import OF_SELF from pretix.helpers.models import modelcopy from pretix.helpers.periodic import minimum_interval from pretix.testutils.middleware import debugflags_var @@ -3508,88 +3513,40 @@ def signal_listener_issue_media(sender: Event, order: Order, **kwargs): ) +def position_not_used_cancellation_check(order: Order, keep: PositionSet, position: OrderPosition): + for pos in order.all_positions.all(): + if pos == position and position not in keep: + for checkin in pos.all_checkins.all(): + if checkin.successful and checkin.list.consider_tickets_used: + return CheckResult( + id="pretixbase_position_not_used", + reason=f"Position used in Checkin {checkin}", + cancellation_possible=False, + ) + else: + return CheckResult( + id="pretixbase_position_not_used", + reason="Position not up for cancellation", + cancellation_possible=True, + ) + + return CheckResult( + id="pretixbase_position_not_used", + reason="Ticket not used", + cancellation_possible=True, + ) - +@receiver(self_service_cancellation_checks, dispatch_uid="pretixbase_position_not_used") +def signal_listener_position_not_used(sender: Event, **kwargs): + return CancellationCheck(id="pretixbase_position_not_used", + type=CheckTypes.POSITION, + check_fn=position_not_used_cancellation_check, + prefetches=[ + lambda: Prefetch('all_positions__all_checkins__list', ) + ]) # TODO weitere System Checks # OrderPositions mit Item.min_per_order dürfen nur storniert werden, wenn genug übrig bleiben oder alle des gleichen Items storniert werden # OrderPositions mit addon_to != None dürfen nur über den bestehenden Add-On-Flow storniert werden # OrderPositions mit is_bundled dürfen nur mit der Parent-Position zusammen storniert werden - -# TODO transaktion -def self_service_cancel(order: Order, keep: Set[OrderPosition], dry_run: bool): - """ - - :param order: - :param keep: - :param dry_run: - :return: - """ - cancellation_checks: List[CancellationCheck] = [resp for recv, resp in self_service_cancellation_checks.send(event=order.event)] - - position_rules = CancellationRule.objects.filter(event=order.event).filter("fee_cancellation_process" == Decimal("0.00")).all() - process_rules = CancellationRule.objects.filter(event=order.event).filter("fee_cancellation_process" != Decimal("0.00")).all() - - # Todo get prefetches/selects from rules as well - prefetches = list(chain.from_iterable([cc.prefetches for cc in cancellation_checks])) - related_selects = list(chain.from_iterable(cc.related_selects for cc in cancellation_checks)) - - per_position_rulings: Dict[int, List[Ruling]] = {} - - prefetched_order = Order.objects.select_related(related_selects).prefetch_related(*prefetches).get(pk=order.pk) - # All queries should be done by now - with ensure_no_queries(): - for position in prefetched_order.positions: - position_rulings = [] - - system_check_results = [cc.check(prefetched_order, keep, position) for cc in cancellation_checks] - - for rule in position_rules: - check_results = [check(prefetched_order, keep, position) for check in rule.checks] - - if rule.fee_percentage_per_item and rule.fee_absolute_per_item: - raise NotImplementedError("Should never be reached") - elif rule.fee_absolute_per_item != Decimal(0.00): - position_rulings.append( - Ruling.from_absolute_fee( - rule_id=rule.id, - results=reduce(lambda a, b: a | b, [*system_check_results, *check_results], {}), - fee_type='position_fee', - absolute_fee=rule.fee_absolute_per_item - ) - ) - else: - position_rulings.append( - Ruling.from_relative_fee( - rule_id=rule.id, - results=reduce(lambda a, b: a | b, [*system_check_results, *check_results], {}), - fee_type='position_fee', - reference_price=position.price, - percentage=rule.fee_absolute_per_item, - currency=order.event.currency - ) - ) - position_rulings.sort() - per_position_rulings[position.id] = position_rulings - effective_position_rulings = [op_rulings[0] for op_rulings in per_position_rulings.values()] - - process_rulings: List[Ruling] = [] - - for rule in process_rules: - check_results = [check(prefetched_order, keep, position) for check in rule.checks] - - process_rulings.append(Ruling.from_absolute_fee( - rule_id=rule.id, - results=reduce(lambda a, b: a | b, [*check_results], {}), - fee_type='process_fee', - absolute_fee=rule.fee_cancellation_process - )) - - process_rulings.sort() - - effective_process_ruling = process_rulings[0] - - cancellation_possible = all([r.cancellation_possible for r in effective_position_rulings]) - - # TODO zusammenführen der Rulings diff --git a/src/pretix/helpers/database.py b/src/pretix/helpers/database.py index 8e75c2aae8..967034bacc 100644 --- a/src/pretix/helpers/database.py +++ b/src/pretix/helpers/database.py @@ -21,6 +21,7 @@ # import contextlib import logging +import os from django.conf import settings from django.core.exceptions import FieldDoesNotExist, ImproperlyConfigured @@ -299,10 +300,9 @@ def ensure_no_queries(): :return: """ def blocker(*args, **kwargs): - if settings.DEBUG: - raise RuntimeError(f"Unexpected DB query: {args[0]}") - else: - logger.error("Unexpected DB query: %s", args[0]) + if settings.DEBUG or "PYTEST_CURRENT_TEST" in os.environ: + raise RuntimeError(f"Unexpected DB query: {args[1]}") + logger.error("Unexpected DB query: %s", args[1]) with connection.execute_wrapper(blocker): yield diff --git a/src/pretix/plugins/ticketoutputpdf/signals.py b/src/pretix/plugins/ticketoutputpdf/signals.py index 3bc2dc131c..198c0f59bc 100644 --- a/src/pretix/plugins/ticketoutputpdf/signals.py +++ b/src/pretix/plugins/ticketoutputpdf/signals.py @@ -28,9 +28,10 @@ from django.utils.translation import gettext_lazy as _ from pretix.base.logentrytypes import EventLogEntryType, log_entry_types from pretix.base.models import Event, SalesChannel -from pretix.base.signals import ( # NOQA: legacy import - EventPluginSignal, event_copy_data, item_copy_data, layout_text_variables, - logentry_display, logentry_object_link, register_data_exporters, +from pretix.base.signals import EventPluginSignal # NOQA: legacy import +from pretix.base.signals import ( + event_copy_data, item_copy_data, layout_text_variables, logentry_display, + logentry_object_link, register_data_exporters, register_multievent_data_exporters, register_ticket_outputs, ) from pretix.control.signals import item_forms, order_position_buttons @@ -38,9 +39,8 @@ from pretix.plugins.ticketoutputpdf.forms import TicketLayoutItemForm from pretix.plugins.ticketoutputpdf.models import ( TicketLayout, TicketLayoutItem, ) -from pretix.presale.style import ( # NOQA: legacy import - get_fonts, register_event_fonts, register_fonts, -) +from pretix.presale.style import get_fonts # NOQA: legacy import +from pretix.presale.style import register_event_fonts, register_fonts @receiver(register_ticket_outputs, dispatch_uid="output_pdf") diff --git a/src/tests/base/test_self_service_cancellation.py b/src/tests/base/test_self_service_cancellation.py index 6c0caee8b6..aa190214e0 100644 --- a/src/tests/base/test_self_service_cancellation.py +++ b/src/tests/base/test_self_service_cancellation.py @@ -1,25 +1,98 @@ +import contextlib +from datetime import timedelta from decimal import Decimal +from typing import cast import pytest +from django.db.models import Prefetch +from django.utils.timezone import now +from django_scopes import scope, scopes_disabled -from pretix.base.models.cancellation import CheckResult, FeeType, RuleResult +from pretix.base.models import ( + Checkin, Event, Order, OrderPosition, Organizer, +) +from pretix.base.models.cancellation import ( + CancellationCheck, CancellationRule, CheckResult, Checks, CheckTypes, + FeeType, PositionResult, ProcessResult, RuleResult, +) +from pretix.base.services.orders import signal_listener_position_not_used +from pretix.helpers import ensure_no_queries -def make_check(possible: bool, *, id: str = "chk", reason: str = "") -> CheckResult: +@pytest.fixture +def event(): + o = Organizer.objects.create(name='Dummy', slug='dummy') + event = Event.objects.create( + organizer=o, name='Dummy', slug='dummy', + date_from=now() + ) + return event + + +@pytest.fixture +def item(event): + return event.items.create( + name='Ticket', + category=None, default_price=23, + admission=True + ) + + +@pytest.fixture +def checkin_list(event): + return event.checkin_lists.create(name="foo", consider_tickets_used=True) + + +@pytest.fixture +def order(event, item): + o = Order.objects.create( + code='123456', event=event, email='dummy@dummy.test', + status=Order.STATUS_PENDING, + datetime=now(), expires=now() + timedelta(days=10), + sales_channel=event.organizer.sales_channels.get(identifier="web"), + total=14, locale='en' + ) + return o + + +@pytest.fixture +def order_position(item, order): + op = OrderPosition.objects.create( + order=order, + item=item, + variation=None, + price=Decimal("14"), + ) + return op + + +def make_check_result(possible: bool, *, id: str = "chk", reason: str = "") -> CheckResult: return CheckResult(id=id, reason=reason, cancellation_possible=possible) -def make_rule(fee, *, possible: bool = True, fee_type: FeeType = FeeType.POSITION, id: int = 1) -> RuleResult: - # ``cancellation_possible`` is derived from the partial check results, so we - # attach a single passing/failing check to control it deterministically. +def make_rule_result(fee, *, possible: bool = True, fee_type: FeeType = FeeType.POSITION, id: int = 1) -> RuleResult: return RuleResult( id=id, - partial_results=[make_check(possible)], + partial_results=[make_check_result(possible)], fee_type=fee_type, fee=Decimal(fee), ) +def make_cancellation_check(id: str, type: CheckTypes, result: bool, prefetches=[], + related_selects=[]) -> CancellationCheck: + def position_check_fn(order, keep, position): + return make_check_result(result, id=id) + + def check_fn(order, keep): + return make_check_result(result, id=id) + + if type == CheckTypes.POSITION: + return CancellationCheck(id, type, position_check_fn, prefetches=prefetches, related_selects=related_selects) + else: + return CancellationCheck(id, type, check_fn, prefetches=prefetches, related_selects=related_selects) + + @pytest.mark.parametrize("partial_results,expected", [ ([True], True), ([False], False), @@ -28,7 +101,7 @@ def make_rule(fee, *, possible: bool = True, fee_type: FeeType = FeeType.POSITIO ([True, False], False), ]) def test_rule_result_cancellation_possible(partial_results, expected): - check_results = [make_check(state) for state in partial_results] + check_results = [make_check_result(state) for state in partial_results] result = RuleResult(id=1, partial_results=check_results, fee_type=FeeType.POSITION, fee=Decimal(0)) assert result.cancellation_possible == expected @@ -37,12 +110,12 @@ def test_rule_result_cancellation_possible(partial_results, expected): @pytest.mark.parametrize( ("left", "right", "expected"), [ - (("5", True), ("10", True), True), # cheaper fee is "less" - (("10", True), ("5", True), False), # pricier fee is not "less" - (("5", True), ("5", True), False), # equal fee is not "less" - (("5", False), ("10", False), True), # cheaper is "less" when both impossible - (("100", True), ("1", False), True), # possible ranks below impossible... - (("1", False), ("100", True), False), # ...and impossible never below possible + (("5", True), ("10", True), True), + (("10", True), ("5", True), False), + (("5", True), ("5", True), False), + (("5", False), ("10", False), True), + (("100", True), ("1", False), True), + (("1", False), ("100", True), False), ], ids=[ "cheaper-lt-pricier-both-possible", @@ -54,6 +127,169 @@ def test_rule_result_cancellation_possible(partial_results, expected): ], ) def test_lt_returns_expected(left, right, expected): - a = make_rule(left[0], possible=left[1]) - b = make_rule(right[0], possible=right[1]) + a = make_rule_result(left[0], possible=left[1]) + b = make_rule_result(right[0], possible=right[1]) assert (a < b) is expected + + +@pytest.mark.parametrize( + ("fee_type", "absolute", "reference", "result"), + [ + (FeeType.MINIMUM, Decimal(10), Decimal(1), Decimal(9)), + (FeeType.MINIMUM, Decimal(1), Decimal(10), Decimal(0)), + (FeeType.MINIMUM, Decimal(10), Decimal(10), Decimal(0)), + (FeeType.ADDITIONAL, Decimal(10), Decimal(1), Decimal(10)), + (FeeType.ADDITIONAL, Decimal(1), Decimal(10), Decimal(1)), + (FeeType.ADDITIONAL, Decimal(10), Decimal(10), Decimal(10)), + ], + ids=[ + "minimum-absolute-less-than-reference", + "minimum-absolute-more-than-reference", + "minimum-absolute-equal-reference", + "additional-absolute-less-than-reference", + "additional-absolute-more-than-reference", + "additional-absolute-equal-reference", + ], +) +def test_from_process_fee(fee_type: FeeType, absolute: Decimal, reference: Decimal, result: Decimal): + res = RuleResult.from_process_fee(id=1, partial_results=[], + fee_type=fee_type, absolute_fee=absolute, reference_price=reference) + assert res.fee == result + + +@pytest.mark.parametrize( + ("check_results", "rule_results", "cancellation_possible", "fee"), + [ + ({1: [make_check_result(True)]}, {1: [make_rule_result(Decimal(10), possible=True)]}, True, Decimal(10)), + ({1: [make_check_result(False)]}, {1: [make_rule_result(Decimal(10), possible=True)]}, False, Decimal(10)), + ({1: [make_check_result(False)]}, {1: [make_rule_result(Decimal(10), possible=False)]}, False, Decimal(0)), + ({1: [make_check_result(True), make_check_result(False)]}, {1: [make_rule_result(Decimal(10), possible=True)]}, + False, Decimal(10)), + ({1: [make_check_result(True)]}, + {1: [make_rule_result(Decimal(10), possible=False), make_rule_result(Decimal(5), possible=True)]}, True, + Decimal(5)), + ({1: [make_check_result(True)]}, + {1: [make_rule_result(Decimal(10), possible=False), make_rule_result(Decimal(5), possible=False)]}, False, + Decimal(0)), + + ], +) +def test_position_results(check_results, rule_results, cancellation_possible, fee): + pos_res = PositionResult(position_check_results=check_results, position_rule_results=rule_results, ) + assert pos_res.cancellation_possible == cancellation_possible + assert pos_res.fee_value == fee + + +@pytest.mark.parametrize( + ("check_results", "rule_results", "cancellation_possible", "fee"), + [ + ([make_check_result(True)], [make_rule_result(Decimal(10), possible=True)], True, Decimal(10)), + ([make_check_result(False)], [make_rule_result(Decimal(10), possible=True)], False, Decimal(10)), + ([make_check_result(False)], [make_rule_result(Decimal(10), possible=False)], False, Decimal(0)), + ([make_check_result(True), make_check_result(False)], [make_rule_result(Decimal(10), possible=True)], + False, Decimal(10)), + ([make_check_result(True)], + [make_rule_result(Decimal(10), possible=False), make_rule_result(Decimal(5), possible=True)], True, + Decimal(5)), + ([make_check_result(True)], + [make_rule_result(Decimal(10), possible=False), make_rule_result(Decimal(5), possible=False)], False, + Decimal(0)), + + ], +) +def test_process_results(check_results, rule_results, cancellation_possible, fee): + pos_res = ProcessResult(process_check_results=check_results, process_rule_results=rule_results, ) + assert pos_res.cancellation_possible == cancellation_possible + assert pos_res.fee_value == fee + + +@pytest.mark.parametrize( + ("received", "position_checks", "process_checks", "raises"), + [ + ([('', make_cancellation_check('pos-1', CheckTypes.POSITION, True))], + [make_cancellation_check('pos-1', CheckTypes.POSITION, True)], + [], + contextlib.nullcontext()), + ([('', make_cancellation_check('proc-1', CheckTypes.PROCESS, True))], + [], + [make_cancellation_check('proc-1', CheckTypes.PROCESS, True)], + contextlib.nullcontext()), + ([('', make_cancellation_check('proc-1', CheckTypes.PROCESS, True)), + ('', make_cancellation_check('proc-1', CheckTypes.PROCESS, True))], + [], + [make_cancellation_check('proc-1', CheckTypes.PROCESS, True)], + pytest.raises(ValueError)), + ([('', make_cancellation_check('pos-1', CheckTypes.POSITION, True)), + ('', make_cancellation_check('pos-1', CheckTypes.POSITION, True))], + [], + [make_cancellation_check('pos-1', CheckTypes.POSITION, True)], + pytest.raises(ValueError)), + ([('', 1)], + [], + [make_cancellation_check('pos-1', CheckTypes.POSITION, True)], + pytest.raises(ValueError)), + ] +) +def test_cancellation_rule_collect_checks(received, position_checks, process_checks, raises): + event = cast(Event, cast(object, {})) + + def send_fn(_event): + return received + + with raises: + checks = CancellationRule._collect_checks(event=event, send_fn=send_fn) + assert checks.position == position_checks + assert checks.process == process_checks + + +@pytest.mark.django_db +def test_prefetch_empty(event, order): + checks = Checks(position=[], process=[]) + with scope(organizer=event.organizer): + prefetched_order = CancellationRule._prefetch_order(event, order, checks) + assert prefetched_order.id == order.id + + +@pytest.mark.django_db +def test_prefetch_incl_values_select_related(event, order): + checks = Checks( + position=[ + make_cancellation_check('pos_1', CheckTypes.POSITION, True, prefetches=[lambda: Prefetch('all_positions')], + related_selects=['organizer'])], + process=[ + make_cancellation_check('proc_1', CheckTypes.PROCESS, True, prefetches=[lambda: Prefetch('all_positions')], + related_selects=['organizer'])] + ) + + with scope(organizer=event.organizer): + prefetched_order = CancellationRule._prefetch_order(event, order, checks) + assert prefetched_order.id == order.id + + +@pytest.mark.django_db +def test_ticket_not_used(event, order, order_position, checkin_list, settings): + position_not_used_check = signal_listener_position_not_used(event) + checks = Checks(position=[position_not_used_check], process=[]) + keep = set() + + with scope(organizer=event.organizer): + prefetched_order = CancellationRule._prefetch_order(event, order, checks) + + with ensure_no_queries(): + result = position_not_used_check.evaluate(prefetched_order, keep, order_position) + + assert result.cancellation_possible is True + + with scope(organizer=event.organizer): + Checkin.objects.create( + list=checkin_list, + position=order_position, + successful=True + ) + + prefetched_order = CancellationRule._prefetch_order(event, order, checks) + with scopes_disabled(): + with ensure_no_queries(): + result = position_not_used_check.evaluate(prefetched_order, keep, order_position) + + assert result.cancellation_possible is False diff --git a/src/tests/helpers/test_database.py b/src/tests/helpers/test_database.py new file mode 100644 index 0000000000..8ca23a84c8 --- /dev/null +++ b/src/tests/helpers/test_database.py @@ -0,0 +1,77 @@ +""" +Tests for the ``ensure_no_queries`` context manager. + +No mocking is used: every test runs against a real database connection via the +``django_db`` marker. ``settings.DEBUG`` is toggled through pytest-django's +``settings`` fixture, which restores the original value after each test. +""" + +import logging + +import pytest +from django_scopes import scopes_disabled + +from pretix.base.models import Event +from pretix.helpers.database import ensure_no_queries + + +@pytest.mark.django_db +def test_raises_runtime_error_in_debug(settings): + settings.DEBUG = True + + with pytest.raises(RuntimeError, match="Unexpected DB query"): + with scopes_disabled(): + with ensure_no_queries(): + Event.objects.exists() + + +@pytest.mark.django_db +def test_logs_error_when_not_debug(settings, caplog): + settings.DEBUG = False + + with caplog.at_level(logging.ERROR): + with scopes_disabled(): + with ensure_no_queries(): + Event.objects.exists() + + assert any( + record.levelno == logging.ERROR + and "Unexpected DB query" in record.getMessage() + for record in caplog.records + ) + + +@pytest.mark.django_db +def test_no_error_without_queries(settings, caplog): + settings.DEBUG = True + + with caplog.at_level(logging.ERROR): + with ensure_no_queries(): + result = sum(range(10)) # pure Python, no DB access + + assert result == 45 + assert "Unexpected DB query" not in caplog.text + + +@pytest.mark.django_db +def test_queries_allowed_after_context(settings): + settings.DEBUG = True + + with ensure_no_queries(): + pass + + with scopes_disabled(): + assert Event.objects.count() == 0 + + +@pytest.mark.django_db +def test_blocker_removed_even_after_exception(settings): + settings.DEBUG = True + + with pytest.raises(RuntimeError, match="Unexpected DB query"): + with scopes_disabled(): + with ensure_no_queries(): + Event.objects.exists() + + with scopes_disabled(): + assert Event.objects.count() == 0