replace user_cancel_allowed internals

This commit is contained in:
Lukas Bockstaller
2026-09-01 17:25:16 +02:00
parent f5ad09233c
commit 14688c5c79
6 changed files with 339 additions and 171 deletions
+69 -67
View File
@@ -90,7 +90,6 @@ class CheckResult:
return cls(**data)
@dataclass(frozen=True)
class RuleResult:
"""
@@ -201,12 +200,12 @@ PositionSet: TypeAlias = Set[OrderPosition]
class PositionCheckFn(Protocol):
def __call__(self, order: Order, keep: PositionSet, position: OrderPosition, check_ts: datetime.datetime,
/) -> CheckResult:
/) -> Optional[CheckResult]:
...
class ProcessCheckFn(Protocol):
def __call__(self, order: Order, keep: PositionSet, check_ts: datetime.datetime, /) -> CheckResult:
def __call__(self, order: Order, keep: PositionSet, check_ts: datetime.datetime, /) -> Optional[CheckResult]:
...
@@ -219,7 +218,7 @@ class CancellationCheck:
related_selects: List[str] = field(default_factory=list)
def evaluate(self, order: Order, keep: PositionSet,
position: OrderPosition | None, check_ts: datetime.datetime) -> CheckResult:
position: OrderPosition | None, check_ts: datetime.datetime) -> Optional[CheckResult]:
if position and self.type == CheckTypes.POSITION:
return self.check_fn(order, keep, position, check_ts)
elif position is None and self.type == CheckTypes.PROCESS:
@@ -320,11 +319,13 @@ class Cancellation(models.Model):
CREATED: Final = "CREATED"
APPROVAL_PENDING: Final = "APPROVAL_PENDING"
PERFORMED: Final = "PERFORMED"
CANCELLED: Final = "CANCELLED"
CANCELLATION_STATE = (
(CREATED, _("Created")),
(APPROVAL_PENDING, _("Approval pending")),
(PERFORMED, _("Performed")),
(CANCELLED, _("Cancelled")),
)
event = models.ForeignKey(
@@ -348,7 +349,7 @@ class Cancellation(models.Model):
auto_now_add=True,
)
cancellation_state = models.CharField(
state = models.CharField(
max_length=16,
choices=CANCELLATION_STATE,
default=CREATED,
@@ -375,7 +376,7 @@ class Cancellation(models.Model):
@staticmethod
def evaluate(event: Event, order: Order, keep: Set[OrderPosition],
check_ts: datetime.datetime) -> "Cancellation":
check_ts: datetime.datetime) -> "CancellationResult":
# validate that all keep order positions are part of the order
for p in keep:
@@ -411,8 +412,9 @@ class Cancellation(models.Model):
# 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, check_ts=check_ts))
res = check.evaluate(order=order, keep=keep, position=position, check_ts=check_ts)
if res is not None:
position_check_results[position.id].append(res)
# evaluate all customer specified rules for this position
for rule in position_rules:
@@ -432,7 +434,9 @@ class Cancellation(models.Model):
# 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, check_ts=check_ts))
res = check.evaluate(order=order, keep=keep, position=None, check_ts=check_ts)
if res is not None:
process_check_results.append(res)
# evaluate all customer specified rules for the cancellation process
for rule in process_rules:
@@ -445,17 +449,17 @@ class Cancellation(models.Model):
res = CancellationResult(position_result=position_results, process_result=process_result)
return res
@staticmethod
def prepare(event: Event, order: Order, keep: Set[OrderPosition],
check_ts: datetime.datetime) -> "Cancellation":
res = Cancellation.evaluate(event=event, order=order, keep=set(), check_ts=check_ts)
c = Cancellation(event=event, order=order, result=res, evaluation_ts=check_ts)
c.save()
c.keep.add(*keep)
return c
def prepare(self):
# TODO: store the cancellation id in the session storage
pass
def execute(self):
# TODO load the cancellation verdict from the id and perform the actions
pass
@@ -480,6 +484,7 @@ class CancellationRuleManager(models.Manager.from_queryset(CancellationRuleQuery
def get_queryset(self):
return super().get_queryset().filter(type=self.check_type).order_by("pk")
class CancellationRule(models.Model):
EARLIEST: Final = "EARLIEST"
LATEST: Final = "LATEST"
@@ -573,10 +578,11 @@ class CancellationRule(models.Model):
),
]
@staticmethod
def collect_checks(event: Event, send_fn: Callable[
[Event], List[Tuple[Any, Any]]] = _send_self_service_cancellation_checks) -> Checks:
[Event], List[Tuple[Any, Any]]
] = _send_self_service_cancellation_checks) -> Checks:
position_checks: List[CancellationCheck] = []
process_checks: List[CancellationCheck] = []
@@ -638,7 +644,6 @@ class CancellationRule(models.Model):
return date_field.datetime(resolve_subevent(reldate_type))
def clean(self):
super().clean()
errors = {}
@@ -675,8 +680,6 @@ class PositionCancellationRuleManager(CancellationRuleManager):
check_type = CheckTypes.POSITION
class PositionCancellationRule(CancellationRule):
"""
PositionCancellationRules answer the questions:
@@ -734,40 +737,39 @@ class PositionCancellationRule(CancellationRule):
)
def _evaluate_cancellation_moment(self, position: OrderPosition, check_ts: datetime.datetime) -> List[CheckResult]:
with ensure_no_queries():
check_results = []
check_results = []
order = position.order
order = position.order
for param in ('allowed_until', 'except_after'):
value: RelativeDateWrapper | None = getattr(self, param, None)
if value is not None:
if check_ts <= self._resolve_date_field(value, order, position):
check_results.append(
CheckResult(
id=f"position_rule_{self.id}_{param}",
reason=_("{} is earlier than {} cutoff {}".format(check_ts, param, value)),
cancellation_possible=True
)
)
else:
check_results.append(
CheckResult(
id=f"position_rule_{self.id}_{param}",
reason=_("{} is later than {} cutoff {}".format(check_ts, param, value)),
cancellation_possible=False
)
for param in ('allowed_until', 'except_after'):
value: RelativeDateWrapper | None = getattr(self, param, None)
if value is not None:
if check_ts <= self._resolve_date_field(value, order, position):
check_results.append(
CheckResult(
id=f"position_rule_{self.id}_{param}",
reason=_("{} is earlier than {} cutoff {}".format(check_ts, param, value)),
cancellation_possible=True
)
)
else:
check_results.append(
CheckResult(
id=f"position_rule_{self.id}_{param}",
reason=_("No {} limit defined".format(param)),
cancellation_possible=True
reason=_("{} is later than {} cutoff {}".format(check_ts, param, value)),
cancellation_possible=False
)
)
else:
check_results.append(
CheckResult(
id=f"position_rule_{self.id}_{param}",
reason=_("No {} limit defined".format(param)),
cancellation_possible=True
)
)
return check_results
return check_results
def evaluate_position_rule(self, order: Order, _keep: Set[OrderPosition], position: OrderPosition,
check_ts: datetime.datetime) -> Optional[RuleResult]:
@@ -854,37 +856,37 @@ class ProcessCancellationRule(CancellationRule):
return CancellationRule._resolve_date_field_common(date_field, order, resolve_subevent)
def _evaluate_cancellation_moment(self, order: Order, check_ts: datetime.datetime) -> List[CheckResult]:
with ensure_no_queries():
check_results: List[CheckResult] = []
for param in ('allowed_until', 'except_after'):
value: RelativeDateWrapper | None = getattr(self, param, None)
if value is not None:
if check_ts <= self._resolve_date_field(value, order, self.subevent_variant):
check_results.append(
CheckResult(
id=f"process_rule_{self.id}_{param}",
reason=_("{} is earlier than {} cutoff {}".format(check_ts, param, value)),
cancellation_possible=True
)
)
else:
check_results.append(
CheckResult(
id=f"process_rule_{self.id}_{param}",
reason=_("{} is later than {} cutoff {}".format(check_ts, param, value)),
cancellation_possible=False
)
check_results: List[CheckResult] = []
for param in ('allowed_until', 'except_after'):
value: RelativeDateWrapper | None = getattr(self, param, None)
if value is not None:
if check_ts <= self._resolve_date_field(value, order, self.subevent_variant):
check_results.append(
CheckResult(
id=f"process_rule_{self.id}_{param}",
reason=_("{} is earlier than {} cutoff {}".format(check_ts, param, value)),
cancellation_possible=True
)
)
else:
check_results.append(
CheckResult(
id=f"process_rule_{self.id}_{param}",
reason=_("No {} limit defined".format(param)),
cancellation_possible=True
reason=_("{} is later than {} cutoff {}".format(check_ts, param, value)),
cancellation_possible=False
)
)
return check_results
else:
check_results.append(
CheckResult(
id=f"process_rule_{self.id}_{param}",
reason=_("No {} limit defined".format(param)),
cancellation_possible=True
)
)
return check_results
def evaluate_process_rule(self, order: Order, _keep: Set[OrderPosition], position_fees: Decimal,
check_ts: datetime.datetime) -> Optional[RuleResult]:
+5 -30
View File
@@ -677,6 +677,7 @@ class Order(LockModel, LoggedModel):
return self.total - self.tax_total
def cancel_allowed(self):
# TODO turn into check after replacing cancellation machinery
return (
self.status in (Order.STATUS_PENDING, Order.STATUS_PAID, Order.STATUS_EXPIRED) and self.count_positions
)
@@ -786,36 +787,10 @@ class Order(LockModel, LoggedModel):
"""
Returns whether or not this order can be canceled by the user.
"""
from .checkin import Checkin
if self.cancellation_requests.exists() or not self.cancel_allowed():
return False
positions = list(
self.positions.all().annotate(
has_checkin=Exists(Checkin.objects.filter(position_id=OuterRef('pk'), list__consider_tickets_used=True))
).select_related('item').prefetch_related('issued_gift_cards')
)
cancelable = all([op.item.allow_cancel and not op.has_checkin and not op.blocked for op in positions])
if not cancelable or not positions:
return False
for op in positions:
for gc in op.issued_gift_cards.all():
if gc.value != op.price:
return False
if op.granted_memberships.with_usages().filter(usages__gt=0):
return False
if self.user_cancel_deadline and time_machine_now() > self.user_cancel_deadline:
return False
if self.status == Order.STATUS_PAID:
if self.total == Decimal('0.00'):
return self.event.settings.cancel_allow_user
return self.event.settings.cancel_allow_user_paid
elif self.payment_refund_sum > Decimal('0.00'):
return False
elif self.status == Order.STATUS_PENDING:
return self.event.settings.cancel_allow_user
return False
from pretix.base.models.cancellation import Cancellation
res = Cancellation.evaluate(event=self.event, order=self, keep=set(),
check_ts=datetime.now(tz=ZoneInfo(self.event.settings.timezone)))
return res.cancellation_possible
def propose_auto_refunds(self, amount: Decimal, payments: list=None):
# Algorithm to choose which payments are to be refunded to create the least hassle
+202 -4
View File
@@ -50,8 +50,8 @@ from django.core.cache import cache
from django.core.exceptions import ValidationError
from django.db import models, transaction
from django.db.models import (
Count, Exists, F, IntegerField, Max, Min, OuterRef, Prefetch, Q, QuerySet, Subquery,
Sum, Value,
Count, Exists, F, IntegerField, Max, Min, OuterRef, Prefetch, Q, QuerySet,
Subquery, Sum, Value,
)
from django.db.models.functions import Cast, Greatest
from django.db.transaction import get_connection
@@ -71,7 +71,9 @@ from pretix.base.models import (
Membership, Order, OrderPayment, OrderPosition, Quota, Seat,
SeatCategoryMapping, User, Voucher,
)
from pretix.base.models.cancellation import CancellationCheck, CheckResult, CheckTypes, PositionSet
from pretix.base.models.cancellation import (
Cancellation, CancellationCheck, CheckResult, CheckTypes, PositionSet,
)
from pretix.base.models.event import Event_SettingsStore, SubEvent
from pretix.base.models.orders import (
BlockedTicketSecret, CheckoutSession, InvoiceAddress, OrderFee,
@@ -3600,7 +3602,7 @@ def signal_listener_issue_media(sender: Event, order: Order, **kwargs):
def position_not_used_cancellation_check(order: Order, keep: PositionSet, position: OrderPosition,
check_ts: datetime):
for pos in order.all_positions.all():
for pos in order.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:
@@ -3632,6 +3634,202 @@ def signal_listener_position_not_used(sender: Event, **kwargs):
lambda: Prefetch('all_positions__all_checkins__list', )
])
def position_not_blocked_check(order: Order, keep: PositionSet, position: OrderPosition,
check_ts: datetime):
for pos in order.positions.all():
if pos == position and position not in keep:
return CheckResult(
id="pretixbase_position_not_blocked",
reason="Position is blocked" if pos.blocked else "Position is not blocked",
cancellation_possible=not pos.blocked,
)
return None
@receiver(self_service_cancellation_checks, dispatch_uid="pretixbase_position_not_blocked")
def signal_listener_position_not_blocked(sender: Event, **kwargs):
return CancellationCheck(id="pretixbase_position_not_blocked",
type=CheckTypes.POSITION,
check_fn=position_not_blocked_check,
prefetches=[
lambda: Prefetch('all_positions', )
])
def position_giftcard_not_used(order: Order, keep: PositionSet, position: OrderPosition,
check_ts: datetime):
for pos in order.positions.all():
if pos == position and position not in keep:
if len(pos.issued_gift_cards.all()) > 0:
for gc in pos.issued_gift_cards.all():
if gc.value != pos.price:
return CheckResult(
id="pretixbase_position_giftcard_not_used",
reason="Issued giftcard was used",
cancellation_possible=False,
)
return CheckResult(
id="pretixbase_position_giftcard_not_used",
reason="Issued giftcard was not used",
cancellation_possible=True,
)
return None
@receiver(self_service_cancellation_checks, dispatch_uid="pretixbase_position_giftcard_not_used")
def signal_listener_position_giftcard_not_used(sender: Event, **kwargs):
return CancellationCheck(id="pretixbase_position_giftcard_not_used",
type=CheckTypes.POSITION,
check_fn=position_giftcard_not_used,
prefetches=[
lambda: Prefetch('all_positions__issued_gift_cards', )
])
def position_membership_not_used(order: Order, keep: PositionSet, position: OrderPosition,
check_ts: datetime):
for pos in order.positions.all():
if pos == position and position not in keep:
if len(pos.granted_memberships.all()) > 0:
for membership in pos.granted_memberships.all():
if membership.usages > 0:
return CheckResult(
id="pretixbase_position_membership_not_used",
reason="Membership was already used",
cancellation_possible=False,
)
return CheckResult(
id="pretixbase_position_giftcard_not_used",
reason="Included Membership was not used",
cancellation_possible=True,
)
else:
return CheckResult(
id="pretixbase_position_membership_not_used",
reason="No membership included",
cancellation_possible=True,
)
return None
@receiver(self_service_cancellation_checks, dispatch_uid="pretixbase_position_membership_not_used")
def signal_position_membership_not_used(sender: Event, **kwargs):
return CancellationCheck(id="pretixbase_position_membership_not_used",
type=CheckTypes.POSITION,
check_fn=position_membership_not_used,
prefetches=[
lambda: Prefetch('all_positions__granted_memberships',
queryset=Membership.objects.with_usages(), )
])
def position_item_allow_cancel(order: Order, keep: PositionSet, position: OrderPosition,
check_ts: datetime):
for pos in order.positions.all():
if pos == position and position not in keep:
return CheckResult(
id="pretixbase_position_item_allow_cancel",
reason="Item can be canceled" if pos.item.allow_cancel else "Item cannot be canceled",
cancellation_possible=pos.item.allow_cancel,
)
return None
@receiver(self_service_cancellation_checks, dispatch_uid="pretixbase_position_item_allow_cancel")
def signal_position_item_allow_cancel(sender: Event, **kwargs):
return CancellationCheck(id="pretixbase_position_item_allow_cancel",
type=CheckTypes.POSITION,
check_fn=position_item_allow_cancel,
prefetches=[
lambda: Prefetch('all_positions__item')
])
def process_order_payment_state(order: Order, keep: PositionSet, check_ts: datetime):
if order.status == Order.STATUS_PAID:
if order.total == Decimal("0.00"):
return CheckResult(
id="pretixbase_process_order_paid",
reason="Free orders can be canceled" if order.event.settings.cancel_allow_user else "Free orders cannot be canceled",
cancellation_possible=order.event.settings.cancel_allow_user,
)
return CheckResult(
id="pretixbase_process_order_paid",
reason="Paid orders can be canceled" if order.event.settings.cancel_allow_user_paid else "Paid orders cannot be canceled",
cancellation_possible=order.event.settings.cancel_allow_user_paid,
)
elif order.payment_refund_sum > Decimal('0.00'):
return CheckResult(
id="pretixbase_process_order_paid",
reason="Outstanding refund sum prevents further cancellations",
cancellation_possible=False,
)
elif order.status == Order.STATUS_PENDING:
return CheckResult(
id="pretixbase_process_order_paid",
reason="Pending orders can be canceled" if order.event.settings.cancel_allow_user else "Pending orders cannot be canceled",
cancellation_possible=order.event.settings.cancel_allow_user,
)
else:
return CheckResult(
id="pretixbase_process_order_paid",
reason="Order is in a state that does not allow cancellations",
cancellation_possible=False,
)
@receiver(self_service_cancellation_checks, dispatch_uid="pretixbase_process_order_payment_state")
def signal_listener_process_order_payment_state(sender: Event, **kwargs):
return CancellationCheck(id="pretixbase_process_order_payment_state",
type=CheckTypes.PROCESS,
check_fn=process_order_payment_state,
prefetches=[]
)
def process_cancel_allowed(order: Order, keep: PositionSet, check_ts: datetime):
cancel_allowed = order.cancel_allowed()
return CheckResult(
id="pretixbase_process_cancel_allowed",
reason="Order allows cancellation" if cancel_allowed else "Order doesn't allow for cancellation",
cancellation_possible=cancel_allowed,
)
@receiver(self_service_cancellation_checks, dispatch_uid="pretixbase_process_cancel_allowed")
def signal_listener_process_cancel_allowed(sender: Event, **kwargs):
return CancellationCheck(id="pretixbase_process_process_cancel_allowed",
type=CheckTypes.PROCESS,
check_fn=process_cancel_allowed,
prefetches=[]
)
def process_cancellation_in_progress(order: Order, keep: PositionSet, check_ts: datetime):
if any([c.state not in [Cancellation.PERFORMED, Cancellation.CANCELLED] for c in order.cancellations.all()]):
return CheckResult(
id="pretixbase_process_cancellation_in_progress",
reason="Cancellation request already in progress",
cancellation_possible=False,
)
return CheckResult(
id="pretixbase_process_cancellation_in_progress",
reason="No cancellation request in progress",
cancellation_possible=True,
)
@receiver(self_service_cancellation_checks, dispatch_uid="pretixbase_process_cancellation_in_progress")
def signal_listener_process_cancellation_in_progress(sender: Event, **kwargs):
return CancellationCheck(id="pretixbase_process_process_cancellation_in_progress",
type=CheckTypes.PROCESS,
check_fn=process_cancellation_in_progress,
prefetches=[]
)
# 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
+1 -1
View File
@@ -301,7 +301,7 @@ def ensure_no_queries():
:return:
"""
def blocker(*args, **kwargs):
if settings.DEBUG or "PYTEST_CURRENT_TEST" in os.environ and not "ENSURE_NO_QUERIES_OVERRIDE" in os.environ:
if settings.DEBUG or "PYTEST_CURRENT_TEST" in os.environ and "ENSURE_NO_QUERIES_OVERRIDE" not in os.environ:
raise RuntimeError(f"Unexpected DB query: {args[1]}")
logger.error("Unexpected DB query: %s", args[1])
+27 -21
View File
@@ -57,6 +57,9 @@ from pretix.base.models import (
Organizer, Question, Quota, ScheduledEventExport, SeatingPlan, User,
Voucher, WaitingListEntry,
)
from pretix.base.models.cancellation import (
CheckTypes, FeeType, ProcessCancellationRule,
)
from pretix.base.models.event import SubEvent
from pretix.base.models.items import (
ItemBundle, SubEventItem, SubEventItemVariation,
@@ -1561,18 +1564,17 @@ class OrderTestCase(BaseQuotaTestCase):
@classscope(attr='o')
def test_user_cancel_absolute_deadline_unpaid_no_subevents(self):
assert self.order.user_cancel_deadline is None
self.event.settings.set('cancel_allow_user_until', RelativeDateWrapper(
now() + timedelta(days=1)
))
r = ProcessCancellationRule(event=self.event, type=CheckTypes.PROCESS,
allowed_until=RelativeDateWrapper(now() + timedelta(days=1)),
fee_mode=FeeType.MINIMUM, fee_cancellation_process=Decimal("0.00"))
self.order = Order.objects.get(pk=self.order.pk)
assert self.order.user_cancel_deadline > now()
assert self.order.user_cancel_allowed
self.event.settings.set('cancel_allow_user_until', RelativeDateWrapper(
now() - timedelta(days=1)
))
r.allowed_until = RelativeDateWrapper(now() - timedelta(days=1))
r.save()
self.order = Order.objects.get(pk=self.order.pk)
assert self.order.user_cancel_deadline < now()
assert not self.order.user_cancel_allowed
@classscope(attr='o')
@@ -1580,18 +1582,19 @@ class OrderTestCase(BaseQuotaTestCase):
self.event.date_from = now() + timedelta(days=3)
self.event.save()
assert self.order.user_cancel_deadline is None
self.event.settings.set('cancel_allow_user_until', RelativeDateWrapper(
r = ProcessCancellationRule(event=self.event, type=CheckTypes.PROCESS, allowed_until=RelativeDateWrapper(
RelativeDate(days=2, time=datetime.time(14, 0, 0), base_date_name='event__date_from', minutes=None)
))
), fee_mode=FeeType.MINIMUM, fee_cancellation_process=Decimal("0.00"))
self.order = Order.objects.get(pk=self.order.pk)
assert self.order.user_cancel_deadline > now()
assert self.order.user_cancel_allowed
self.event.settings.set('cancel_allow_user_until', RelativeDateWrapper(
RelativeDate(days=4, time=datetime.time(14, 0, 0), base_date_name='event__date_from', minutes=None)
))
r.allowed_until = RelativeDateWrapper(
RelativeDate(days=4, time=datetime.time(14, 0, 0), base_date_name='event__date_from',
minutes=None))
r.save()
self.order = Order.objects.get(pk=self.order.pk)
assert self.order.user_cancel_deadline < now()
assert not self.order.user_cancel_allowed
@classscope(attr='o')
@@ -1606,15 +1609,18 @@ class OrderTestCase(BaseQuotaTestCase):
self.op2.subevent = se2
self.op2.save()
self.event.settings.set('cancel_allow_user_until', RelativeDateWrapper(
r = ProcessCancellationRule(event=self.event, allowed_until=RelativeDateWrapper(
RelativeDate(days=2, time=datetime.time(14, 0, 0), base_date_name='event__date_from', minutes=None)
))
), fee_mode=FeeType.MINIMUM, fee_cancellation_process=Decimal("0.00"))
r.save()
self.order = Order.objects.get(pk=self.order.pk)
assert self.order.user_cancel_deadline < now()
assert not self.order.user_cancel_allowed
self.op2.subevent = se1
self.op2.save()
self.order = Order.objects.get(pk=self.order.pk)
assert self.order.user_cancel_deadline > now()
assert self.order.user_cancel_allowed
@classscope(attr='o')
def test_user_cancel_fee(self):
@@ -230,6 +230,7 @@ class TestPositionResult:
assert pos_res.cancellation_possible is True
assert pos_res.fee_value == Decimal(10)
class TestProcessResults:
@pytest.mark.parametrize(
("check_results", "rule_results", "cancellation_possible", "fee"),
@@ -245,7 +246,6 @@ class TestProcessResults:
([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(self, check_results, rule_results, cancellation_possible, fee):
@@ -279,45 +279,33 @@ class TestCancellationRule:
@pytest.mark.parametrize(
("received", "position_checks", "process_checks", "raises"),
[
(
[make_cancellation_check('pos-1', CheckTypes.POSITION, True)],
[0],
[],
contextlib.nullcontext()
),
(
[make_cancellation_check('proc-1', CheckTypes.PROCESS, True)],
[],
[0],
contextlib.nullcontext()
),
(
[make_cancellation_check('pos-1', CheckTypes.POSITION, True),
make_cancellation_check('proc-1', CheckTypes.PROCESS, True)],
[0],
[1],
contextlib.nullcontext()
),
(
[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)],
[],
[],
pytest.raises(ValueError)
),
(
[('', 1)],
[],
[],
pytest.raises(ValueError)
),
([make_cancellation_check('pos-1', CheckTypes.POSITION, True)],
[0],
[],
contextlib.nullcontext()),
([make_cancellation_check('proc-1', CheckTypes.PROCESS, True)],
[],
[0],
contextlib.nullcontext()),
([make_cancellation_check('pos-1', CheckTypes.POSITION, True),
make_cancellation_check('proc-1', CheckTypes.PROCESS, True)],
[0],
[1],
contextlib.nullcontext()),
([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)],
[],
[],
pytest.raises(ValueError)),
([('', 1)],
[],
[],
pytest.raises(ValueError)),
]
)
def test_cancellation_rule_collect_checks(self, received, position_checks, process_checks, raises):
@@ -666,7 +654,6 @@ class TestCancellationRule:
else:
raise ValueError("Unknown cancellation rule type: {}".format(rule_type_variants))
setattr(r, attr, RelativeDateWrapper(reference_ts))
r.save()
rule = rule_object.objects.get(id=r.id)
@@ -710,14 +697,14 @@ class TestCancellationRule:
with scope(organizer=event.organizer):
cancellation = Cancellation.evaluate(event, order, keep=set(), check_ts=check_ts)
assert cancellation.possible == True
assert cancellation.cancellation_possible is True
position_rule_results = cancellation.result.position_result.position_rule_results[1]
position_rule_results = cancellation.position_result.position_rule_results[1]
assert len(position_rule_results) == 2
assert position_rule_results[0].cancellation_possible == True
assert position_rule_results[1].cancellation_possible == False
assert position_rule_results[0].cancellation_possible is True
assert position_rule_results[1].cancellation_possible is False
process_rule_results = cancellation.result.process_result.process_rule_results
process_rule_results = cancellation.process_result.process_rule_results
assert len(process_rule_results) == 2
assert process_rule_results[0].cancellation_possible == True
assert process_rule_results[1].cancellation_possible == False
assert process_rule_results[0].cancellation_possible is True
assert process_rule_results[1].cancellation_possible is False