CGM_Public/pretix_original
2
0
Fork 1
mirror of https://github.com/pretix/pretix.git synced 2026-08-07 10:17:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

prefetching and documenting stuff

Browse Source
This commit is contained in:
Lukas Bockstaller
2026-07-09 17:15:11 +02:00
parent a6b5cef3b6
commit cc280d5f6c
3 changed files with 215 additions and 368 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pretix/base/models/cancellation.py
+166 -91
View File
@@ -1,7 +1,8 @@
from abc import ABC
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Dict, List, Literal, NamedTuple, Optional, Protocol, Set, TYPE_CHECKING, TypeAlias
from itertools import chain
from typing import Callable, Dict, List, Literal, Optional, Protocol, Set, TYPE_CHECKING, TypeAlias
from django.core.exceptions import ValidationError
from django.core.validators import MaxValueValidator, MinValueValidator
@@ -14,32 +15,82 @@ from pretix.base.models import Event, Item, ItemVariation, Order, OrderPosition
from pretix.base.reldate import ModelRelativeDateTimeField
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
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.
Number 2 is trickier because organizers will have complex^(TM) cancellation
fee structures and expressing these in an understandable way is a challenge.
Especially when taking support cases into consideration that have to debug
certain behaviour long after.
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
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
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
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.
"""
class FeeType(models.TextChoices):
"""
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")
ADDITIONAL = "add_process_fee", _("Additional fee")
POSITION = "position_fee", _("Position fee")
class RuleTypes(models.TextChoices):
class CheckTypes(models.TextChoices):
POSITION = "position", _("Order Position Cancellation Rule")
PROCESS = "process", _("Cancellation Process Rule")
@dataclass(frozen=True)
class CheckResult:
"""
Result of an individual cancellation check.
The check result only encodes if the check allows or disallows cancellation via
`cancellation_possible`
"""
id: str
reason: str
cancellation_possible: bool
type: Literal['check'] = field(default="check")
@property
def key(self) -> str:
return f"{self.type}::{self.id}"
@dataclass(frozen=True)
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
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
- minimum process fees, the total cancellation fee across all positions and the process must be at least this
- additional process fee, an additional processing fee is charged in addition to the per position fees
"""
id: int
partial_results: List[CheckResult]
fee_type: FeeType
@@ -47,10 +98,6 @@ class RuleResult:
type: Literal['rule'] = field(default="rule")
@property
def key(self) -> str:
return f"{self.type}::{self.id}"
@property
def cancellation_possible(self) -> bool:
return all(result.cancellation_possible for result in self.partial_results)
@@ -102,27 +149,30 @@ class RuleResult:
if not isinstance(other, RuleResult):
return NotImplemented
if self.fee_type != other.fee_type:
return NotImplemented
if self.cancellation_possible == other.cancellation_possible:
return self.fee < other.fee
else:
return self.cancellation_possible and not other.cancellation_possible
class Checks(NamedTuple):
@dataclass(frozen=True)
class Checks:
position: List["CancellationCheck"]
process: List["CancellationCheck"]
@property
def prefetches(self) -> List[Callable[[], Prefetch]]:
return list(chain([check.prefetches for check in [*self.position, *self.process]]))
@property
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: ...
class ProcessCheckFn(Protocol):
def __call__(self, order: Order, keep: PositionSet) -> CheckResult: ...
@@ -130,21 +180,73 @@ class ProcessCheckFn(Protocol):
@dataclass(frozen=True)
class CancellationCheck(ABC):
id: str
type: RuleTypes
type: CheckTypes
check_fn: PositionCheckFn | ProcessCheckFn
prefetches: List[Prefetch] = field(default_factory=list)
prefetches: List[Callable[[], Prefetch]] = field(default_factory=list)
related_selects: List[str] = field(default_factory=list)
def evaluate(self, order: Order, keep: PositionSet,
position: OrderPosition | None) -> CheckResult:
if position and self.type == RuleTypes.POSITION:
if position and self.type == CheckTypes.POSITION:
return self.check_fn(order, keep, position)
elif position is None and self.type == RuleTypes.PROCESS:
elif position is None and self.type == CheckTypes.PROCESS:
return self.check_fn(order, keep)
else:
raise ValidationError("Type of the rule doesn't match the check_fn")
@dataclass(frozen=True)
class PositionResult:
position_check_results: Dict[int, List[CheckResult]]
position_rule_results: Dict[int, List[RuleResult]]
def cancellation_possible(self) -> bool:
def ok(results: list) -> bool:
return results[0].cancellation_possible if results else True
return all(
ok(results)
for d in (self.position_check_results, self.position_rule_results)
for results in d.values()
)
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]
if best_option.cancellation_possible:
fee_value += best_option.fee
return fee_value
@dataclass(frozen=True)
class ProcessResult:
process_check_results: List[CheckResult]
process_rule_results: List[RuleResult]
def cancellation_possible(self) -> bool:
return all([res.cancellation_possible for res in [*self.process_check_results, *self.process_rule_results]])
@dataclass(frozen=True)
class CancellationResult:
position_result: PositionResult
process_result: ProcessResult
def cancellation_possible(self) -> bool:
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
pass
def perform_cancellation(self, order: Order, keep: Set[int]):
# TODO load the cancellation verdict from the session and perform the actions
pass
class CancellationRule(models.Model):
event = models.ForeignKey(
Event,
@@ -155,18 +257,22 @@ class CancellationRule(models.Model):
type = models.CharField(
verbose_name=_("Type of the cancellation rule"),
default=RuleTypes.POSITION,
choices=RuleTypes,
default=CheckTypes.POSITION,
choices=CheckTypes,
max_length=15,
)
allowed_until = ModelRelativeDateTimeField(null=True, blank=True)
except_after = ModelRelativeDateTimeField(null=True, blank=True)
prefetches: List[Callable[[], Prefetch]] = []
related_selects: List[str] = []
@staticmethod
def _collect_checks(event: Event) -> 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):
@@ -175,33 +281,35 @@ class CancellationRule(models.Model):
raise ValueError('self_service_cancellation_checks received multiple responses with the id')
seen.add(resp.id)
if resp.type == RuleTypes.POSITION:
if resp.type == CheckTypes.POSITION:
position_checks.append(resp)
if resp.type == RuleTypes.PROCESS:
if resp.type == CheckTypes.PROCESS:
process_checks.append(resp)
return Checks(position=position_checks, process=process_checks)
@staticmethod
def evaluate(event: Event, order: Order, positions_to_keep: Set[int]):
# collect all position checks and all process_checks
def evaluate(event: Event, order: Order, keep: Set[OrderPosition]) -> "CancellationResult":
# collect all checks, position_rules and process_rules that are applicable
checks = CancellationRule._collect_checks(event=event)
position_rules = PositionCancellationRule.objects.filter(event=event, type=CheckTypes.POSITION)
process_rules = ProcessCancellationRule.objects.filter(event=event, type=CheckTypes.PROCESS)
# TODO prefetch the order
# TODO set keep to Set[OrderPosition]
keep: Set[OrderPosition] = set()
position_rules = PositionCancellationRule.objects.filter(event=event, type=RuleTypes.POSITION)
process_rules = ProcessCancellationRule.objects.filter(event=event, type=RuleTypes.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)
# keep track of all decisions so we can explain them in the logs
position_check_results: Dict[int, List[CheckResult]] = {}
position_rule_results: Dict[int, List[RuleResult]] = {}
process_check_results: List[CheckResult] = []
process_rule_results: List[RuleResult] = []
total_pos_fees = Decimal(0)
# perform position checks
# perform all position checks and position rules
for position in order.positions.all():
position_check_results[position.id] = []
position_rule_results[position.id] = []
@@ -210,7 +318,7 @@ class CancellationRule(models.Model):
if position.id in keep:
continue
# evaluate the system provided system checks for the position
# evaluate the system/plugin checks for the position
for check in checks.position:
position_check_results[position.id].append(check.evaluate(order=order, keep=keep, position=position))
@@ -220,24 +328,30 @@ class CancellationRule(models.Model):
if result is not None:
position_rule_results[position.id].append(result)
# get the cheapest rulings and sum up their fees
position_rule_results[position.id].sort()
best_option = position_rule_results[position.id][0]
if best_option.cancellation_possible:
total_pos_fees += best_option.fee
position_results = PositionResult(position_check_results=position_check_results,
position_rule_results=position_rule_results)
# evaluate all system provided checks for the cancellation process
# we need the current fee_value to select the cheapest process rule
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] = []
process_rule_results: List[RuleResult] = []
# evaluate all system/plugin provided checks for the cancellation process
for check in checks.process:
process_check_results.append(check.evaluate(order=order, keep=keep, position=None))
# evaluate all customer specified rules for the cancellation process
for rule in process_rules:
result = rule.evaluate_process_rule(order=order, keep=keep, position_fees=total_pos_fees)
result = rule.evaluate_process_rule(order=order, keep=keep, position_fees=temp_position_fees)
if result is not None:
process_rule_results.append(result)
process_rule_results.sort()
return CancellationResult(position_check_results=position_check_results, pos)
process_result = ProcessResult(process_check_results=process_check_results,
process_rule_results=process_rule_results)
return CancellationResult(position_result=position_results, process_result=process_result)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -276,6 +390,9 @@ class PositionCancellationRule(CancellationRule):
ItemVariation, blank=True, verbose_name=_("Variations")
)
prefetches: List[Callable[[], Prefetch]] = []
related_selects: List[str] = []
if TYPE_CHECKING:
allowed_until = ModelRelativeDateTimeField(null=True, blank=True)
except_after = ModelRelativeDateTimeField(null=True, blank=True)
@@ -337,6 +454,9 @@ class ProcessCancellationRule(CancellationRule):
max_length=15,
)
prefetches: List[Callable[[], Prefetch]] = []
related_selects: List[str] = []
if TYPE_CHECKING:
allowed_until = ModelRelativeDateTimeField(null=True, blank=True)
except_after = ModelRelativeDateTimeField(null=True, blank=True)
@@ -357,48 +477,3 @@ class ProcessCancellationRule(CancellationRule):
absolute_fee=self.fee_cancellation_process,
reference_price=position_fees,
)
@dataclass(frozen=True)
class CancellationResult:
position_check_results: Dict[int, List[CheckResult]]
position_rule_results: Dict[int, List[RuleResult]]
process_check_results: List[CheckResult]
process_rule_results: List[RuleResult]
def _position_checks_passed(self) -> bool:
passed: List[bool] = []
for pos, results in self.position_check_results.items():
# customer did not wish to cancel this position
if len(results) == 0:
passed += True
else:
passed += results[0].cancellation_possible
return all(passed)
def _position_rules_passed(self) -> bool:
passed: List[bool] = []
for pos, results in self.position_rule_results.items():
# customer did not wish to cancel this position
if len(results) == 0:
passed += True
else:
passed += results[0].cancellation_possible
return all(passed)
def _process_checks_passed(self) -> bool:
return all([res.cancellation_possible for res in self.process_check_results])
def _process_rules_passed(self) -> bool:
return all([res.cancellation_possible for res in self.process_rule_results])
def cancellation_possible(self) -> bool:
return self._position_checks_passed() and self._position_rules_passed() and self._process_checks_passed() and self._process_rules_passed()
def remember_cancellation(self):
# TODO: store the cancellation verdict in the session storage for X Minutes
pass
def perform_cancellation(self, order: Order, keep: Set[int]):
# TODO load the cancellation verdict from the session and perform the actions
pass
src/pretix/base/services/orders.py
+4 -75
View File
@@ -50,11 +50,7 @@ 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, Prefetch, Q, QuerySet,
Sum, Value,
)
from django.db.models import Prefetch
from django.db.models import Count, Exists, F, IntegerField, Max, Min, OuterRef, 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,21 +65,18 @@ from pretix.base.email import get_email_context
from pretix.base.i18n import get_language_without_region, language
from pretix.base.media import MEDIA_TYPES
from pretix.base.models import (
CartPosition, Checkin, Device, Event, GiftCard, Item, ItemVariation,
CartPosition, Device, Event, GiftCard, Item, ItemVariation,
Membership, Order, OrderPayment, OrderPosition, Quota, Seat,
SeatCategoryMapping, User, Voucher,
)
from pretix.base.models.cancellation import (
CancellationCheckResult, CancellationCheckResultsById, CancellationRule,
Ruling, CancellationCheck
)
from pretix.base.models.cancellation import (CancellationCheck, CancellationRule)
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, TaxedPrice, TaxRule
from pretix.base.models.tax import TAXED_ZERO, TaxRule, TaxedPrice
from pretix.base.payment import GiftCardPayment, PaymentException
from pretix.base.reldate import RelativeDateWrapper
from pretix.base.secrets import assign_ticket_secret
@@ -3515,72 +3508,8 @@ def signal_listener_issue_media(sender: Event, order: Order, **kwargs):
)
class OrderPositionNotUsedCheck(CancellationCheck):
id = "SYSTEM_TICKET_NOT_USED"
prefetches = [
Prefetch(
'checkins',
queryset=Checkin.objects.filter(list__consider_tickets_used=True),
to_attr='used_checkins' # stores result in a list attribute
)
]
related_selects = []
def check(self, order: Order, keep: Set[OrderPosition], order_position: OrderPosition) -> CancellationCheckResultsById:
if order_position.checkins.filter(list__consider_tickets_used=True).exists():
return {self.id: CancellationCheckResult(
cancellation_possible=False,
reason="Order position was used",
)}
else:
return {self.id: CancellationCheckResult(
cancellation_possible=True,
reason="Order position not yet used",
)}
@receiver(self_service_cancellation_checks, dispatch_uid="pretixbase_not_used")
def cancellation_checks_not_used(sender: Event):
return OrderPositionNotUsedCheck()
class NotDiscountedCheck(CancellationCheck):
"""
Check that ensures that orders containing discounted order_positions cannot
be canceled partially.
This is a stop-gap solution until the `discount_grouper` attribute for
AbstractPositions is introduced, allowing us to be more grannular
"""
id = "SYSTEM_NO_DISCOUNTED_ORDER_POSITIONS"
prefetches = [
]
related_selects = []
def check(self, order: Order, keep: Set[OrderPosition], order_position: OrderPosition) -> CancellationCheckResultsById:
cancellations = Set(order.positions).difference(keep)
if order_position in cancellations:
if order_position.discount_id is None:
return {self.id: CancellationCheckResult(
cancellation_possible=True,
reason=_("Order position was bought without discount"),
)}
else:
return {self.id: CancellationCheckResult(
cancellation_possible=False,
reason=_("Order position was bought with a discount"),
)}
else:
return {self.id: CancellationCheckResult(
cancellation_possible=False,
reason=_("Order position not canceled - check not applicable"),
)}
@receiver(self_service_cancellation_checks, dispatch_uid="pretixbase_not_discountend")
def cancellation_checks_not_discounted(sender: Event):
return NotDiscountedCheck()
# TODO weitere System Checks
src/tests/base/test_self_service_cancellation.py
+45 -202
View File
@@ -1,216 +1,59 @@
from datetime import date, datetime, timedelta
from decimal import Decimal
from zoneinfo import ZoneInfo
import pytest
from django.utils.timezone import make_aware, now
from django_scopes import scope
from freezegun import freeze_time
from pretix.base.models import Event, Item, Order, OrderPosition, Organizer
from pretix.base.models.cancellation import CancellationRule, OrderDiff
NOW = now()
DAYS_UNTIL_EVENT=60
EVENT_START = NOW+timedelta(days=DAYS_UNTIL_EVENT)
from pretix.base.models.cancellation import CheckResult, FeeType, RuleResult
def make_check(possible: bool, *, id: str = "chk", reason: str = "") -> CheckResult:
return CheckResult(id=id, reason=reason, cancellation_possible=possible)
@pytest.fixture()
def event():
o = Organizer.objects.create(name='Dummy', slug='dummy', plugins='pretix.plugins.banktransfer')
event = Event.objects.create(
organizer=o, name='Dummy', slug='dummy',
date_from=EVENT_START,
plugins='pretix.plugins.banktransfer'
)
return event
@pytest.fixture()
def item1(event):
return Item.objects.create(event=event, name='Early-bird item1',
default_price=Decimal('23.00'), admission=True)
@pytest.fixture()
def order(event):
return Order.objects.create(
code='FOO', event=event, email='dummy@dummy.test',
status=Order.STATUS_PENDING, locale='en',
datetime=NOW,
total=0,
sales_channel=event.organizer.sales_channels.get(identifier="web"),
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.
return RuleResult(
id=id,
partial_results=[make_check(possible)],
fee_type=fee_type,
fee=Decimal(fee),
)
@pytest.mark.django_db
def test_status_rule(event, item1, order):
with scope(organizer=event.organizer, event=event):
op = OrderPosition.objects.create(
order=order, item=item1, variation=None,
price=Decimal("0.00"), attendee_name_parts={'full_name': "Peter"}, positionid=1
)
@pytest.mark.parametrize("partial_results,expected", [
([True], True),
([False], False),
([True, True], True),
([False, False], False),
([True, False], False),
])
def test_rule_result_cancellation_possible(partial_results, expected):
check_results = [make_check(state) for state in partial_results]
cancellation_rule = CancellationRule.objects.create(
organizer=event.organizer, event=event,
allowed_if_in_order_status=Order.STATUS_PENDING
)
cancellation_rule.items.set([item1])
diff = OrderDiff.cancel_all(order)
assert cancellation_rule._check_order_status(diff=diff, order_position=op) == {
'ORDER_STATUS': CheckRes(
cancellation_possible=True,
reason="Order in required status: 'n'",
),
}
cancellation_rule = CancellationRule.objects.create(
organizer=event.organizer, event=event,
allowed_if_in_order_status=Order.STATUS_PAID
)
cancellation_rule.items.set([item1])
assert cancellation_rule._check_order_status(diff=diff, order_position=op) == {
'ORDER_STATUS': CheckRes(
cancellation_possible=False,
reason="Order in status 'n' cannot be canceled",
),
}
result = RuleResult(id=1, partial_results=check_results, fee_type=FeeType.POSITION, fee=Decimal(0))
assert result.cancellation_possible == expected
@pytest.mark.django_db
def test_timing(event, item1, order):
with scope(organizer=event.organizer, event=event):
order.status = Order.STATUS_PAID
order.save()
OrderPosition.objects.create(
order=order, item=item1, variation=None,
price=Decimal("0.00"), attendee_name_parts={'full_name': "Peter"}, positionid=1
)
cr1 = CancellationRule.objects.create(
organizer=event.organizer, event=event,
allowed_until=now() + timedelta(hours=1),
)
cr1.items.set([item1])
diff = OrderDiff.cancel_all(order)
with freeze_time(now()):
possible, verdicts = CancellationRule.objects.all().cancellation_possible(diff)
assert possible == True
with freeze_time(now()+timedelta(hours=2)):
possible, verdicts=CancellationRule.objects.all().cancellation_possible(diff)
assert possible == False
@pytest.mark.django_db
def test_multiple_limits(event, item1, order):
with (scope(organizer=event.organizer, event=event)):
order.status = Order.STATUS_PAID
order.save()
OrderPosition.objects.create(
order=order, item=item1, variation=None,
price=Decimal("100.00"), attendee_name_parts={'full_name': "Peter"}, positionid=1
)
# free in the first hour after booking
cr1=CancellationRule.objects.create(
organizer=event.organizer, event=event,
allowed_until=NOW + timedelta(hours=1),
)
cr1.items.set([item1])
# free until 30 days before event
cr2 = CancellationRule.objects.create(
organizer=event.organizer, event=event,
allowed_until=EVENT_START - timedelta(days=30),
)
cr2.items.set([item1])
# 50% until 14 days before event
cr3 = CancellationRule.objects.create(
organizer=event.organizer, event=event,
allowed_until=EVENT_START - timedelta(days=14),
fee_percentage_per_item=Decimal(50.0)
)
cr3.items.set([item1])
# 80% until 7 days before event
cr4 = CancellationRule.objects.create(
organizer=event.organizer, event=event,
allowed_until=EVENT_START - timedelta(days=7),
fee_percentage_per_item=Decimal(80.0)
)
cr4.items.set([item1])
# 100% until 1 day before event
cr5 = CancellationRule.objects.create(
organizer=event.organizer, event=event,
allowed_until=EVENT_START - timedelta(days=1),
fee_percentage_per_item=Decimal(100)
)
cr5.items.set([item1])
# Cancellation is not allowed at all, but rule doesn't match the item
CancellationRule.objects.create(
organizer=event.organizer, event=event,
allowed_until=NOW,
fee_percentage_per_item=Decimal(100)
)
diff = OrderDiff.cancel_all(order)
possible_trace = []
cost_trace = []
for days in range(DAYS_UNTIL_EVENT):
today = NOW + timedelta(days=days)
with freeze_time(today):
possible, verdicts=CancellationRule.objects.all().cancellation_possible(
diff)
possible_trace.append(possible)
cost_trace.append(verdicts[0].total_fee)
assert possible_trace == [True] * 59 + [False]
assert cost_trace == [Decimal("0.0000")] * 30 + \
[Decimal("50.0000")] * 16 + \
[Decimal("80.0000")] * 7 + \
[Decimal("100.0000")] * 6 + \
[Decimal("0.0000")]
@pytest.mark.django_db
def test_cancellation_rule_query_set(event, item1, order):
with scope(organizer=event.organizer, event=event):
OrderPosition.objects.create(
order=order, item=item1, variation=None,
price=Decimal("0.00"), attendee_name_parts={'full_name': "Peter"}, positionid=1
)
cr1 = CancellationRule.objects.create(
organizer=event.organizer, event=event,
allowed_if_in_order_status=Order.STATUS_PENDING, fee_absolute_per_order=Decimal('10.00'),
)
cr1.items.set([item1])
cr2 = CancellationRule.objects.create(
organizer=event.organizer, event=event,
allowed_if_in_order_status=Order.STATUS_PAID
)
cr2.items.set([item1])
diff = OrderDiff.cancel_all(order)
possible, verdicts = CancellationRule.objects.all().cancellation_possible(diff)
assert possible == True
@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
],
ids=[
"cheaper-lt-pricier-both-possible",
"pricier-not-lt-cheaper-both-possible",
"equal-fee-not-lt",
"cheaper-lt-pricier-both-impossible",
"possible-lt-impossible-despite-higher-fee",
"impossible-not-lt-possible",
],
)
def test_lt_returns_expected(left, right, expected):
a = make_rule(left[0], possible=left[1])
b = make_rule(right[0], possible=right[1])
assert (a < b) is expected