repair tests

# Conflicts:
#	src/pretix/base/services/orders.py

doc/development/api/general.rst

@@ -15,7 +15,7 @@ Core
       item_copy_data, register_sales_channel_types, register_global_settings, quota_availability, global_email_filter,
       register_ticket_secret_generators, gift_card_transaction_display,
       register_text_placeholders, register_mail_placeholders, device_info_updated,
-      register_event_permission_groups, register_organizer_permission_groups
+      register_event_permission_groups, register_organizer_permission_groups, self_service_cancellation_checks
 
 Order events
 """"""""""""

pyproject.toml

@@ -129,6 +129,7 @@ dev = [
   "pytest==9.1.*",
   "playwright",
   "responses",
+	"django-stubs-ext"
 ]
 
 [project.entry-points."distutils.commands"]

src/pretix/base/models/cancellation.py

@@ -0,0 +1,903 @@
+import datetime
+import operator
+from dataclasses import asdict, dataclass, field
+from decimal import Decimal
+from itertools import chain
+from typing import (
+    TYPE_CHECKING, Any, Callable, ClassVar, Dict, Final, List, Literal,
+    Optional, Protocol, Set, Tuple, TypeAlias,
+)
+
+from django.core.exceptions import ValidationError
+from django.core.serializers.json import DjangoJSONEncoder
+from django.core.validators import MaxValueValidator, MinValueValidator
+from django.db import models
+from django.db.models import Prefetch, QuerySet
+from django.utils.timezone import make_aware
+from django.utils.translation import gettext_lazy as _
+from django_stubs_ext import StrOrPromise
+
+from pretix.base.decimal import round_decimal
+from pretix.base.models import Event, Item, ItemVariation, Order, OrderPosition
+from pretix.base.reldate import ModelRelativeDateTimeField, RelativeDateWrapper
+from pretix.base.signals import self_service_cancellation_checks
+from pretix.helpers import ensure_no_queries
+
+"""
+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 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
+    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 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: StrOrPromise
+    cancellation_possible: bool
+    type: Literal['check'] = field(default="check")
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "CheckResult":
+        return cls(**data)
+
+
+
+@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` 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
+    - 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
+    fee: Decimal
+
+    type: Literal['rule'] = field(default="rule")
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "RuleResult":
+        return cls(
+            id=data["id"],
+            partial_results=[CheckResult.from_dict(r) for r in data["partial_results"]],
+            fee_type=FeeType(data["fee_type"]),
+            fee=Decimal(data["fee"]),
+        )
+
+    @property
+    def cancellation_possible(self) -> bool:
+        return all(result.cancellation_possible for result in self.partial_results)
+
+    @classmethod
+    def from_absolute_fee(
+            cls,
+            id: int,
+            partial_results: List[CheckResult],
+            fee_type: Literal[FeeType.POSITION],
+            absolute_fee: Decimal
+    ) -> "RuleResult":
+        return RuleResult(id=id, partial_results=partial_results, fee_type=fee_type, fee=absolute_fee)
+
+    @classmethod
+    def from_relative_fee(
+            cls,
+            id: int,
+            partial_results: List[CheckResult],
+            fee_type: Literal[FeeType.POSITION],
+            position_price: Decimal,
+            percentage: Decimal,
+            currency: str
+    ) -> "RuleResult":
+        return RuleResult(id=id, partial_results=partial_results, fee_type=fee_type,
+                          fee=round_decimal(position_price * (percentage / 100), currency))
+
+    @classmethod
+    def from_process_fee(
+            cls,
+            id: int,
+            partial_results: List[CheckResult],
+            fee_type: Literal[FeeType.MINIMUM, FeeType.ADDITIONAL],
+            absolute_fee: Decimal,
+            reference_price: Decimal
+    ) -> "RuleResult":
+        if fee_type == FeeType.MINIMUM:
+            if reference_price < absolute_fee:
+                fee = absolute_fee - reference_price
+            else:
+                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: object) -> bool:
+        if not isinstance(other, RuleResult):
+            return NotImplemented
+
+        if self.cancellation_possible == other.cancellation_possible:
+            return self.fee < other.fee
+        else:
+            return self.cancellation_possible and not other.cancellation_possible
+
+
+@dataclass(frozen=True)
+class Checks:
+    position: List["CancellationCheck"]
+    process: List["CancellationCheck"]
+
+    @property
+    def prefetches(self) -> List[Callable[[], Prefetch]]:
+        return list(chain.from_iterable(
+            check.prefetches for check in [*self.position, *self.process]
+        ))
+
+    @property
+    def related_selects(self) -> List[str]:
+        return list(chain.from_iterable(
+            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, check_ts: datetime.datetime,
+                 /) -> CheckResult:
+        ...
+
+
+class ProcessCheckFn(Protocol):
+    def __call__(self, order: Order, keep: PositionSet, check_ts: datetime.datetime, /) -> CheckResult:
+        ...
+
+
+@dataclass(frozen=True)
+class CancellationCheck:
+    id: str
+    type: CheckTypes
+    check_fn: PositionCheckFn | ProcessCheckFn = field(compare=False)
+    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, check_ts: datetime.datetime) -> 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:
+            return self.check_fn(order, keep, check_ts)
+        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]]
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "PositionResult":
+        return cls(
+            position_check_results={
+                int(pos_id): [CheckResult.from_dict(r) for r in results]
+                for pos_id, results in data["position_check_results"].items()
+            },
+            position_rule_results={
+                int(pos_id): [RuleResult.from_dict(r) for r in results]
+                for pos_id, results in data["position_rule_results"].items()
+            },
+        )
+
+    @property
+    def cancellation_possible(self) -> bool:
+        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,
+             {key: [min(pos_res)] for key, pos_res in self.position_rule_results.items() if pos_res})
+            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:
+                best_option = min(results)
+                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]
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "ProcessResult":
+        return cls(
+            process_check_results=[CheckResult.from_dict(r) for r in data["process_check_results"]],
+            process_rule_results=[RuleResult.from_dict(r) for r in data["process_rule_results"]],
+        )
+
+    @property
+    def cancellation_possible(self) -> bool:
+        results: List[CheckResult | RuleResult] = [*self.process_check_results]
+        if self.process_rule_results:
+            results.append(min(self.process_rule_results))
+        return all(res.cancellation_possible for res in results)
+
+    @property
+    def fee_value(self) -> Decimal:
+        if not self.process_rule_results:
+            return Decimal("0.00")
+        best_option = min(self.process_rule_results)
+        if best_option.cancellation_possible:
+            return best_option.fee
+        return Decimal("0.00")
+
+
+@dataclass(frozen=True)
+class CancellationResult:
+    position_result: PositionResult
+    process_result: ProcessResult
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "CancellationResult":
+        return cls(
+            position_result=PositionResult.from_dict(data["position_result"]),
+            process_result=ProcessResult.from_dict(data["process_result"]),
+        )
+
+    @property
+    def cancellation_possible(self) -> bool:
+        return self.position_result.cancellation_possible and self.process_result.cancellation_possible
+
+
+class Cancellation(models.Model):
+    CREATED: Final = "CREATED"
+    APPROVAL_PENDING: Final = "APPROVAL_PENDING"
+    PERFORMED: Final = "PERFORMED"
+
+    CANCELLATION_STATE = (
+        (CREATED, _("Created")),
+        (APPROVAL_PENDING, _("Approval pending")),
+        (PERFORMED, _("Performed")),
+    )
+
+    event = models.ForeignKey(
+        Event,
+        verbose_name=_("Event"),
+        related_name="cancellations",
+        on_delete=models.CASCADE
+    )
+    order = models.ForeignKey(
+        Order,
+        verbose_name=_("Order"),
+        related_name="cancellations",
+        on_delete=models.CASCADE
+    )
+    keep = models.ManyToManyField(
+        to=OrderPosition,
+        verbose_name=_("Positions to keep"),
+    )
+    evaluation_ts = models.DateTimeField(
+        verbose_name=_("Cancellation datetime"),
+        auto_now_add=True,
+    )
+
+    cancellation_state = models.CharField(
+        max_length=16,
+        choices=CANCELLATION_STATE,
+        default=CREATED,
+        verbose_name=_("State of the cancellation"),
+    )
+
+    _result = models.JSONField(default=dict, db_column="result", encoder=DjangoJSONEncoder)
+
+    @property
+    def result(self) -> CancellationResult:
+        return CancellationResult.from_dict(self._result)
+
+    @result.setter
+    def result(self, value: CancellationResult):
+        if not isinstance(value, CancellationResult):
+            raise TypeError("result must be a CancellationResult instance")
+        if self._result:
+            raise ValueError("result is write-once and has already been set")
+        self._result = asdict(value)
+
+    @property
+    def possible(self) -> bool:
+        return self.result.cancellation_possible
+
+    @staticmethod
+    def evaluate(event: Event, order: Order, keep: Set[OrderPosition],
+                 check_ts: datetime.datetime) -> "Cancellation":
+
+        # validate that all keep order positions are part of the order
+        for p in keep:
+            if p.order_id != order.id:
+                raise ValidationError("OrderPosition {} does not belong to order {}".format(p.code, order.code))
+
+        # exclude canceled positions
+        for p in order.positions.all():
+            if p.canceled:
+                keep.add(p)
+
+        # collect all checks, position_rules and process_rules that are applicable
+        checks = CancellationRule.collect_checks(event=event)
+        position_rules: QuerySet[PositionCancellationRule] = PositionCancellationRule.objects.filter(
+            event=event).with_rule_data().all()
+        process_rules: QuerySet[ProcessCancellationRule] = ProcessCancellationRule.objects.filter(
+            event=event).with_rule_data().all()
+
+        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]] = {}
+        position_rule_results: Dict[int, List[RuleResult]] = {}
+
+        # perform all position checks and position rules
+        for position in order.positions.all():
+            position_check_results[position.id] = []
+            position_rule_results[position.id] = []
+
+            # skip this position if customer doesn't want to cancel
+            if position in keep:
+                continue
+
+            # 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))
+
+            # evaluate all customer specified rules for this position
+            for rule in position_rules:
+                result = rule.evaluate_position_rule(order, keep, position, check_ts)
+                if result is not None:
+                    position_rule_results[position.id].append(result)
+
+        position_results = PositionResult(position_check_results=position_check_results,
+                                          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
+
+        # 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, check_ts=check_ts))
+
+        # evaluate all customer specified rules for the cancellation process
+        for rule in process_rules:
+            result = rule.evaluate_process_rule(order, keep, temp_position_fees, check_ts)
+            if result is not None:
+                process_rule_results.append(result)
+
+        process_result = ProcessResult(process_check_results=process_check_results,
+                                       process_rule_results=process_rule_results)
+
+        res = CancellationResult(position_result=position_results, process_result=process_result)
+
+        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
+
+
+def _send_self_service_cancellation_checks(event: Event) -> List[Tuple[Any, Any]]:
+    return self_service_cancellation_checks.send(sender=event)
+
+
+class CancellationRuleQuerySet(models.QuerySet):
+    def with_rule_data(self):
+        model = self.model
+        qs = self.prefetch_related(*[p() for p in model.rule_prefetches])
+        if model.rule_related_selects:
+            qs = qs.select_related(*model.rule_related_selects)
+        return qs
+
+
+class CancellationRuleManager(models.Manager.from_queryset(CancellationRuleQuerySet)):
+    check_type: ClassVar[CheckTypes]
+
+    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"
+
+    SUBEVENT_VARIANT_CHOICES = (
+        (EARLIEST, _("Earliest")),
+        (LATEST, _("Latest")),
+    )
+
+    event = models.ForeignKey(
+        Event,
+        verbose_name=_("Event"),
+        related_name="cancellation_rules",
+        on_delete=models.CASCADE
+    )
+
+    type = models.CharField(
+        verbose_name=_("Type of the cancellation rule"),
+        default=CheckTypes.POSITION,
+        choices=CheckTypes,
+        max_length=15,
+    )
+
+    allowed_until = ModelRelativeDateTimeField(null=True, blank=True, verbose_name=_("Allowed until"))
+    except_after = ModelRelativeDateTimeField(null=True, blank=True, verbose_name=_("Except after"))
+    if TYPE_CHECKING:
+        allowed_until: Optional[RelativeDateWrapper]
+        except_after: Optional[RelativeDateWrapper]
+
+    # --- position-only fields ---
+    fee_percentage_per_position = models.DecimalField(
+        max_digits=5,
+        decimal_places=2,
+        validators=[MinValueValidator(Decimal("0.00")), MaxValueValidator(Decimal("100.00"))],
+        verbose_name=_("Fee Percentage per OrderPosition"),
+        default=Decimal("0.00"),
+    )
+    fee_absolute_per_position = models.DecimalField(
+        max_digits=13,
+        decimal_places=2,
+        verbose_name=_("Absolute fee per OrderPosition"),
+        default=Decimal("0.00"),
+        validators=[MinValueValidator(Decimal("0.00"))],
+    )
+
+    all_products = models.BooleanField(
+        verbose_name=_("All products and variations"),
+        default=True,
+    )
+    limit_products = models.ManyToManyField(Item, verbose_name=_("Products"), blank=True)
+    limit_variations = models.ManyToManyField(
+        ItemVariation, blank=True, verbose_name=_("Variations")
+    )
+
+    # --- process-only fields ---
+    subevent_variant = models.CharField(
+        max_length=8,
+        choices=SUBEVENT_VARIANT_CHOICES,
+        default=EARLIEST,
+        verbose_name=_("Subevent variant"),
+        help_text=_("An order can contain tickets for multiple different events if the event has "
+                    "subevents enabled. This choice controls if the order position for the earliest "
+                    "or the latest point in time in the order is used to determine the allowed until and "
+                    "except after dates.")
+    )
+
+    fee_cancellation_process = models.DecimalField(
+        max_digits=13,
+        decimal_places=2,
+        verbose_name=_("Absolute fee per Cancellation"),
+        default=Decimal("0.00"),
+        validators=[MinValueValidator(Decimal("0.00"))],
+    )
+
+    fee_mode = models.CharField(
+        verbose_name=_("The method with which process and position fees are combined."),
+        choices=[
+            (FeeType.MINIMUM, FeeType.MINIMUM.label),
+            (FeeType.ADDITIONAL, FeeType.ADDITIONAL.label),
+        ],
+        blank=True,
+        null=True,
+        max_length=15,
+    )
+
+    class Meta:
+        constraints = [
+            models.CheckConstraint(
+                condition=models.Q(type__in=[CheckTypes.POSITION, CheckTypes.PROCESS]),
+                name="cancellation_rule_type_valid",
+            ),
+        ]
+
+
+    @staticmethod
+    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 send_fn(event):
+            if resp is None:
+                continue
+
+            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')
+            seen.add(resp.id)
+
+            if resp.type == CheckTypes.POSITION:
+                position_checks.append(resp)
+            if resp.type == CheckTypes.PROCESS:
+                process_checks.append(resp)
+
+        return Checks(position=position_checks, process=process_checks)
+
+    @staticmethod
+    def prefetch_order(event: Event, order: Order, checks: Checks) -> Order:
+        prefetches = [pref() for pref in [*checks.prefetches,
+                                          *PositionCancellationRule.prefetches,
+                                          *ProcessCancellationRule.prefetches]]
+
+        related_selects = {*checks.related_selects,
+                           *PositionCancellationRule.related_selects,
+                           *ProcessCancellationRule.related_selects}
+
+        qs = Order.objects.prefetch_related(*prefetches)
+        if related_selects:
+            qs = qs.select_related(*related_selects)
+
+        return qs.get(event=event, id=order.id)
+
+    @staticmethod
+    def _resolve_date_field_common(
+            date_field: RelativeDateWrapper,
+            order: Order,
+            resolve_subevent: Callable[[Any], Any],
+    ) -> datetime.date | datetime.datetime:
+        reldate_type = date_field.choice
+
+        if reldate_type == "date":
+            return make_aware(
+                datetime.datetime.combine(date_field.date(order.event), datetime.time(hour=23, minute=59, second=59)),
+                order.event.timezone,
+            )
+        elif reldate_type == "datetime":
+            return date_field.datetime(order.event)
+
+        if reldate_type.base == "order":
+            return date_field.datetime(order)
+
+        if not order.event.has_subevents:
+            return date_field.datetime(order.event)
+
+        return date_field.datetime(resolve_subevent(reldate_type))
+
+
+    def clean(self):
+        super().clean()
+        errors = {}
+
+        if self.type == CheckTypes.PROCESS:
+            if self.fee_mode not in (FeeType.MINIMUM, FeeType.ADDITIONAL):
+                errors["fee_mode"] = _(
+                    "Fee mode is not valid on a process rule."
+                )
+            if self.fee_percentage_per_position or self.fee_absolute_per_position:
+                errors["fee_percentage_per_position"] = _(
+                    "Position fees must be unset on a process rule."
+                )
+            if self.pk and (self.limit_products.exists() or self.limit_variations.exists()):
+                errors["limit_products"] = _(
+                    "Product/variation limits are not valid on a process rule."
+                )
+
+        if self.type == CheckTypes.POSITION:
+            if self.fee_cancellation_process:
+                errors["fee_cancellation_process"] = _(
+                    "Process fee must be unset on a position rule."
+                )
+            if self.fee_mode:
+                errors["fee_mode"] = _(
+                    "Fee mode is not valid on a position rule."
+                )
+
+        if errors:
+            raise ValidationError(errors)
+
+
+class PositionCancellationRuleManager(CancellationRuleManager):
+    check_type = CheckTypes.POSITION
+
+
+
+
+class PositionCancellationRule(CancellationRule):
+    """
+    PositionCancellationRules answer the questions:
+    - Can this position be canceled?
+    - What is the price for cancelling this position?
+    """
+    objects = PositionCancellationRuleManager()
+
+    rule_prefetches: ClassVar[List[Callable[[], Prefetch]]] = [
+        lambda: Prefetch('limit_products'),
+        lambda: Prefetch('limit_variations'),
+    ]
+    rule_related_selects: ClassVar[List[str]] = []
+
+    prefetches: ClassVar[List[Callable[[], Prefetch]]] = [
+        lambda: Prefetch('all_positions__item'),
+        lambda: Prefetch('event'),
+    ]
+    related_selects: ClassVar[List[str]] = []
+
+    class Meta:
+        proxy = True
+
+    def save(self, *args, **kwargs):
+        self.type = CheckTypes.POSITION
+        self.full_clean()
+        super().save(*args, **kwargs)
+
+    def _position_matches_rule(self, position: OrderPosition) -> Optional[CheckResult]:
+        with ensure_no_queries():
+            res = CheckResult(
+                id=f"position_rule_{self.id}",
+                reason=_("Rule matches this product"),
+                cancellation_possible=True
+            )
+
+            if self.all_products:
+                return res
+
+            item_pks = {item.pk for item in self.limit_products.all()}
+            if position.item_id in item_pks:
+                return res
+
+            variation_pks = {variation.pk for variation in self.limit_variations.all()}
+            if position.variation_id in variation_pks:
+                return res
+
+            return None
+
+    @staticmethod
+    def _resolve_date_field(date_field: RelativeDateWrapper, order: Order,
+                            position: OrderPosition) -> datetime.date | datetime.datetime:
+        return CancellationRule._resolve_date_field_common(
+            date_field, order, resolve_subevent=lambda _reldate_type: position.subevent
+        )
+
+    def _evaluate_cancellation_moment(self, position: OrderPosition, check_ts: datetime.datetime) -> List[CheckResult]:
+        with ensure_no_queries():
+            check_results = []
+
+            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
+                            )
+                        )
+                else:
+                    check_results.append(
+                        CheckResult(
+                            id=f"position_rule_{self.id}_{param}",
+                            reason=_("No {} limit defined".format(param)),
+                            cancellation_possible=True
+                        )
+                    )
+
+            return check_results
+
+    def evaluate_position_rule(self, order: Order, _keep: Set[OrderPosition], position: OrderPosition,
+                               check_ts: datetime.datetime) -> Optional[RuleResult]:
+        rule_check_results = []
+        match = self._position_matches_rule(position)
+        if match:
+            rule_check_results.append(match)
+        rule_check_results.extend(self._evaluate_cancellation_moment(position, check_ts))
+
+        if self.fee_percentage_per_position and self.fee_absolute_per_position:
+            raise NotImplementedError(
+                "Combination of fee_percentage_per position and fee_absolute_per_position is not valid")
+        elif self.fee_absolute_per_position != Decimal(0.00):
+            return RuleResult.from_absolute_fee(
+                id=self.id,
+                partial_results=rule_check_results,
+                fee_type=FeeType.POSITION,
+                absolute_fee=self.fee_absolute_per_position
+            )
+        else:
+            return RuleResult.from_relative_fee(
+                id=self.id,
+                partial_results=rule_check_results,
+                fee_type=FeeType.POSITION,
+                position_price=position.price,
+                percentage=self.fee_percentage_per_position,
+                currency=order.event.currency
+            )
+
+
+class ProcessCancellationRuleManager(CancellationRuleManager):
+    check_type = CheckTypes.PROCESS
+
+
+class ProcessCancellationRule(CancellationRule):
+    """
+    ProcessCancellationRules answer the question:
+    - What is the processing fee for performing this cancellation?
+    """
+
+    objects = ProcessCancellationRuleManager()
+
+    rule_prefetches: ClassVar[List[Callable[[], Prefetch]]] = []
+    rule_related_selects: ClassVar[List[str]] = []
+
+    prefetches: ClassVar[List[Callable[[], Prefetch]]] = [
+        lambda: Prefetch('event'),
+    ]
+    related_selects: ClassVar[List[str]] = []
+
+    class Meta:
+        proxy = True
+
+    def save(self, *args, **kwargs):
+        self.type = CheckTypes.PROCESS
+        self.full_clean()
+        super().save(*args, **kwargs)
+
+    @staticmethod
+    def _resolve_date_field(date_field: RelativeDateWrapper, order: Order,
+                            mode: Literal["EARLIEST", "LATEST"] | str) -> datetime.date | datetime.datetime:
+        if mode not in ('EARLIEST', 'LATEST'):
+            raise ValidationError('Mode is invalid')
+
+        comparators = {
+            "EARLIEST": operator.lt,
+            "LATEST": operator.gt,
+        }
+        compare = comparators[mode]
+
+        def resolve_subevent(reldate_type):
+            base_event = order.event
+            base_value: None | datetime.date = None
+            for pos in order.positions.all():
+                e = pos.subevent if pos.subevent else pos.event
+                value = getattr(e, reldate_type.attribute)
+                if value is None:
+                    continue  # skip when there is no value
+                if base_value is None or compare(value, base_value):
+                    base_event = e
+                    base_value = value
+            return base_event
+
+        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
+                            )
+                        )
+                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]:
+        fee_mode = self.fee_mode
+        if fee_mode not in (FeeType.MINIMUM, FeeType.ADDITIONAL):
+            raise ValueError(f"Unexpected fee_mode: {fee_mode!r}")
+
+        check_results: List[CheckResult] = self._evaluate_cancellation_moment(order, check_ts)
+
+        return RuleResult.from_process_fee(
+            id=self.id,
+            partial_results=check_results,
+            fee_type=fee_mode,
+            absolute_fee=self.fee_cancellation_process,
+            reference_price=position_fees,
+        )

src/pretix/base/reldate.py

@@ -111,21 +111,21 @@ class RelativeDate:
     base_date_name: str = 'event__date_from__'
 
     def __post_init__(self) -> None:
-        if self.is_after and not self._choice.supports_after:
+        if self.is_after and not self.choice.supports_after:
             raise ValueError(
                 "The selected base date and attribute combination does not support relative dates placed after the base date"
             )
-        if not self.is_after and not self._choice.supports_before:
+        if not self.is_after and not self.choice.supports_before:
             raise ValueError(
                 "The selected base date and attribute combination does not support relative dates placed before the base date")
 
     @property
-    def _choice(self):
+    def choice(self):
         return BaseChoice.find(BASE_CHOICES, self.base_date_name)
 
     @property
     def key(self):
-        return self._choice.key
+        return self.choice.key
 
     def __eq__(self, o: object) -> bool:
         if not isinstance(o, RelativeDate):
@@ -141,7 +141,7 @@ class RelativeDate:
         """
         from .models import Event, Order, SubEvent
 
-        choice = self._choice
+        choice = self.choice
 
         if choice.base == "order" and isinstance(base, Order):
             event = base.event
@@ -210,13 +210,13 @@ class RelativeDate:
         if self.minutes is not None:
             return 'RELDATE/minutes/{}/{}/{}'.format(  #
                 self.minutes,
-                self._choice.key,
+                self.choice.key,
                 'after' if self.is_after else '',
             )
         return 'RELDATE/{}/{}/{}/{}'.format(  #
             self.days,
             self.time.strftime('%H:%M:%S') if self.time else '-',
-            self._choice.key,
+            self.choice.key,
             'after' if self.is_after else '',
         )
 
@@ -270,6 +270,15 @@ class RelativeDateWrapper:
     def __init__(self, data: Union[datetime.datetime, RelativeDate]):
         self.data = data
 
+    @property
+    def choice(self) -> Literal["datetime", "date"] | BaseChoice:
+        if isinstance(self.data, datetime.datetime):
+            return "datetime"
+        elif isinstance(self.data, datetime.date):
+            return "date"
+        else:
+            return self.data.choice
+
     def date(self, base: "Event | Order | SubEvent") -> datetime.date:
         """
         If the RelativeDateWrapper wraps a RelativeDate object:

src/pretix/base/services/orders.py

@@ -50,7 +50,7 @@ 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, Q, QuerySet, Subquery,
+    Count, Exists, F, IntegerField, Max, Min, OuterRef, Prefetch, Q, QuerySet, Subquery,
     Sum, Value,
 )
 from django.db.models.functions import Cast, Greatest
@@ -71,6 +71,7 @@ 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.event import Event_SettingsStore, SubEvent
 from pretix.base.models.orders import (
     BlockedTicketSecret, CheckoutSession, InvoiceAddress, OrderFee,
@@ -103,7 +104,7 @@ from pretix.base.signals import (
     order_approved, order_canceled, order_changed, order_denied, order_expired,
     order_expiry_changed, order_fee_calculation, order_paid, order_placed,
     order_reactivated, order_split, order_valid_if_pending, periodic_task,
-    validate_order,
+    self_service_cancellation_checks, validate_order,
 )
 from pretix.base.timemachine import time_machine_now, time_machine_now_assigned
 from pretix.celery_app import app
@@ -1680,7 +1681,8 @@ class OrderChangeManager:
         @property
         def position(self) -> OrderPosition:
             if self._positions is None:
-                raise RuntimeError("Order position has not been created yet. Call commit() first on OrderChangeManager.")
+                raise RuntimeError(
+                    "Order position has not been created yet. Call commit() first on OrderChangeManager.")
             if len(self._positions) != 1:
                 raise RuntimeError("More than one position created.")
             return self._positions[0]
@@ -1899,7 +1901,8 @@ class OrderChangeManager:
 
     def add_position(self, item: Item, variation: ItemVariation, price: Decimal, addon_to: OrderPosition = None,
                      subevent: SubEvent = None, seat: Seat = None, membership: Membership = None,
-                     valid_from: datetime = None, valid_until: datetime = None, count: int = 1) -> 'OrderChangeManager.AddPositionResult':
+                     valid_from: datetime = None, valid_until: datetime = None,
+                     count: int = 1) -> 'OrderChangeManager.AddPositionResult':
         if count < 1:
             raise ValueError("Count must be positive")
         if isinstance(seat, str):
@@ -2616,19 +2619,20 @@ class OrderChangeManager:
                     )
                     nextposid += 1
                     new_pos.append(pos)
-                    new_logs.append(self.order.log_action('pretix.event.order.changed.add', user=self.user, auth=self.auth, data={
-                        'position': pos.pk,
-                        'item': op.item.pk,
-                        'variation': op.variation.pk if op.variation else None,
-                        'addon_to': op.addon_to.pk if op.addon_to else None,
-                        'price': op.price.gross,
-                        'positionid': pos.positionid,
-                        'membership': pos.used_membership_id,
-                        'subevent': op.subevent.pk if op.subevent else None,
-                        'seat': op.seat.pk if op.seat else None,
-                        'valid_from': op.valid_from.isoformat() if op.valid_from else None,
-                        'valid_until': op.valid_until.isoformat() if op.valid_until else None,
-                    }, save=False))
+                    new_logs.append(
+                        self.order.log_action('pretix.event.order.changed.add', user=self.user, auth=self.auth, data={
+                            'position': pos.pk,
+                            'item': op.item.pk,
+                            'variation': op.variation.pk if op.variation else None,
+                            'addon_to': op.addon_to.pk if op.addon_to else None,
+                            'price': op.price.gross,
+                            'positionid': pos.positionid,
+                            'membership': pos.used_membership_id,
+                            'subevent': op.subevent.pk if op.subevent else None,
+                            'seat': op.seat.pk if op.seat else None,
+                            'valid_from': op.valid_from.isoformat() if op.valid_from else None,
+                            'valid_until': op.valid_until.isoformat() if op.valid_until else None,
+                        }, save=False))
 
                 op.result._positions = new_pos
                 LogEntry.bulk_create_and_postprocess(new_logs)
@@ -2956,7 +2960,8 @@ class OrderChangeManager:
         return total
 
     def _check_order_size(self):
-        if (len(self.order.positions.all()) + sum([op.count for op in self._operations if isinstance(op, self.AddOperation)])) > settings.PRETIX_MAX_ORDER_SIZE:
+        if (len(self.order.positions.all()) + sum([op.count for op in self._operations if isinstance(op,
+                                                                                                     self.AddOperation)])) > settings.PRETIX_MAX_ORDER_SIZE:
             raise OrderError(
                 self.error_messages['max_order_size'] % {
                     'max': settings.PRETIX_MAX_ORDER_SIZE,
@@ -3586,3 +3591,43 @@ def signal_listener_issue_media(sender: Event, order: Order, **kwargs):
                         'customer': order.customer_id,
                     }
                 )
+
+
+def position_not_used_cancellation_check(order: Order, keep: PositionSet, position: OrderPosition,
+                                         check_ts: datetime):
+    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

src/pretix/base/signals.py

@@ -1207,3 +1207,11 @@ This signal is sent out each time the information for a Device is modified.
 Both the original and updated versions of the Device are included to allow
 receivers to see what has been updated.
 """
+
+self_service_cancellation_checks = EventPluginSignal()
+"""
+This signal is sent out to collect checks to approve or deny a self service cancellation.
+You are expected to return a class instance that implements CancellationCheck.
+It is is expected that that the CheckFn will not issue any further queries.
+As with all event-plugin signals, the ``sender`` keyword argument will contain the event.
+"""

src/pretix/helpers/database.py

@@ -20,6 +20,8 @@
 # <https://www.gnu.org/licenses/>.
 #
 import contextlib
+import logging
+import os
 
 from django.conf import settings
 from django.contrib.postgres.indexes import BrinIndex
@@ -30,6 +32,8 @@ from django.db.models import (
 )
 from django.utils.functional import lazy
 
+logger = logging.getLogger(__name__)
+
 
 class DummyRollbackException(Exception):
     pass
@@ -288,6 +292,23 @@ def get_deterministic_ordering(model, ordering):
     return ordering
 
 
+@contextlib.contextmanager
+def ensure_no_queries():
+    """
+    Ensures that no database queries are being made in that context.
+    Raises a RuntimeError if running in DEBUG mode, otherwise logs
+    an error.
+    :return:
+    """
+    def blocker(*args, **kwargs):
+        if settings.DEBUG or "PYTEST_CURRENT_TEST" in os.environ and not "ENSURE_NO_QUERIES_OVERRIDE" 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
+
+
 @contextlib.contextmanager
 def conditional_atomic(do_atomic, **kwargs):
     if do_atomic:

src/tests/base/test_self_service_cancellation.py

@@ -0,0 +1,723 @@
+import contextlib
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+from typing import List, Literal, cast
+
+import pytest
+from django.db.models import Prefetch
+from django.utils.timezone import now
+from django_scopes import scope
+
+from pretix.base.models import Checkin, Event, Order, OrderPosition, Organizer
+from pretix.base.models.cancellation import (
+    Cancellation, CancellationCheck, CancellationRule, CheckResult, Checks,
+    CheckTypes, FeeType, PositionCancellationRule, PositionResult,
+    ProcessCancellationRule, ProcessResult, RuleResult,
+)
+from pretix.base.reldate import RelativeDate, RelativeDateWrapper
+from pretix.base.services.orders import signal_listener_position_not_used
+from pretix.helpers import ensure_no_queries
+
+
+@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_result(fee, *, possible: bool = True, fee_type: FeeType = FeeType.POSITION, id: int = 1) -> RuleResult:
+    return RuleResult(
+        id=id,
+        partial_results=[make_check_result(possible)],
+        fee_type=fee_type,
+        fee=Decimal(fee),
+    )
+
+
+def make_cancellation_check(id: str, type: CheckTypes, result: bool, prefetches=None,
+                            related_selects=None,
+                            check_ts: datetime = datetime.now(tz=UTC)) -> CancellationCheck:
+    if related_selects is None:
+        related_selects = []
+    if prefetches is None:
+        prefetches = []
+
+    def position_check_fn(_order, _keep, _position, _check_ts=check_ts):
+        return make_check_result(result, id=id)
+
+    def check_fn(_order, _keep, _check_ts=check_ts):
+        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)
+
+
+class TestRuleResult:
+    @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(self, partial_results: List[bool], expected: bool):
+        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
+
+    @pytest.mark.parametrize(
+        ("left", "right", "expected"),
+        [
+            (("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",
+            "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(self, left, right, expected):
+        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(
+            self,
+            fee_type: Literal[FeeType.MINIMUM, FeeType.ADDITIONAL],
+            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(
+        ("position_price", "percentage", "result"),
+        [
+            (Decimal(10), Decimal(10), Decimal(1)),
+            (Decimal(10), Decimal("9.9"), Decimal("0.99"))
+        ],
+    )
+    def test_from_relative_fee(self, position_price, percentage, result):
+        res = RuleResult.from_relative_fee(id=1,
+                                           partial_results=[],
+                                           fee_type=FeeType.POSITION,
+                                           position_price=position_price,
+                                           percentage=percentage,
+                                           currency="EUR")
+        assert res.fee == result
+
+
+class TestPositionResult:
+    @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(self, 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
+
+    def test_position_with_no_rule_results_does_not_raise(self):
+        # A position that has check results but no matching rule results at all
+        # must not blow up min() on [].
+        check_results = {1: [make_check_result(True)]}
+        rule_results = {1: []}
+
+        pos_res = PositionResult(position_check_results=check_results, position_rule_results=rule_results)
+
+        assert pos_res.cancellation_possible is True
+        assert pos_res.fee_value == Decimal(0)
+
+    def test_mixed_positions_one_without_rule_results(self):
+        # One position has rules, another has none.
+        # The empty one shouldn't crash the overall evaluation or affect the other.
+        check_results = {1: [make_check_result(True)], 2: []}
+        rule_results = {1: [make_rule_result(Decimal(10), possible=True)], 2: []}
+
+        pos_res = PositionResult(position_check_results=check_results, position_rule_results=rule_results)
+
+        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"),
+        [
+            ([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(self, 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
+
+    def test_process_with_no_rules_configured_does_not_raise(self):
+        # No ProcessCancellationRule configured for the event at all: process_rule_results == [].
+        # Should be treated as "no process fee, doesn't block cancellation", not crash.
+        check_results = [make_check_result(True)]
+        rule_results = []
+
+        proc_res = ProcessResult(process_check_results=check_results, process_rule_results=rule_results)
+
+        assert proc_res.cancellation_possible is True
+        assert proc_res.fee_value == Decimal("0.00")
+
+    def test_process_with_no_rules_but_failing_check(self):
+        # Empty rule_results shouldn't mask a failing check-based result.
+        check_results = [make_check_result(False)]
+        rule_results = []
+
+        proc_res = ProcessResult(process_check_results=check_results, process_rule_results=rule_results)
+
+        assert proc_res.cancellation_possible is False
+        assert proc_res.fee_value == Decimal("0.00")
+
+
+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)
+            ),
+        ]
+    )
+    def test_cancellation_rule_collect_checks(self, received, position_checks, process_checks, raises):
+        event = cast(Event, cast(object, {}))
+
+        def send_fn(_event):
+            return [("", res) for res in received]
+
+        with raises:
+            checks = CancellationRule.collect_checks(event=event, send_fn=send_fn)
+
+        for pos in process_checks:
+            assert received[pos] in checks.process
+
+        for pos in position_checks:
+            assert received[pos] in checks.position
+
+    class TestPrefetching:
+
+        @pytest.mark.django_db
+        def test_prefetch_no_checks_collected(self, 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(self, 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
+
+    class TestChecks:
+
+        @pytest.mark.django_db
+        def test_ticket_not_used(self, event, order, order_position, checkin_list):
+            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,
+                                                              datetime.now(tz=UTC))
+                assert result.cancellation_possible is True
+
+                Checkin.objects.create(
+                    list=checkin_list,
+                    position=order_position,
+                    successful=True
+                )
+                prefetched_order = CancellationRule.prefetch_order(event, order, checks)
+
+                with ensure_no_queries():
+                    result = position_not_used_check.evaluate(prefetched_order, keep, order_position,
+                                                              datetime.now(tz=UTC))
+
+                assert result.cancellation_possible is False
+
+    class TestResolveDateFields:
+
+        REFERENCE_DT = datetime(2017, 12, 27, 4, 0, 0, tzinfo=UTC)
+
+        @pytest.fixture(params=["date", "datetime", "order", "event"])
+        def rdt_reldate_variants(self, request):
+            return request.param
+
+        @pytest.fixture
+        def rdt_reldate(self, rdt_reldate_variants) -> RelativeDateWrapper:
+            if rdt_reldate_variants == 'date' or rdt_reldate_variants == 'datetime':
+                return RelativeDateWrapper.from_string(self.REFERENCE_DT.isoformat())
+            elif rdt_reldate_variants == 'order':
+                return RelativeDateWrapper(
+                    RelativeDate(days=1, time=None, base_date_name='order__datetime', minutes=None, is_after=True))
+            elif rdt_reldate_variants == 'event':
+                return RelativeDateWrapper(
+                    RelativeDate(days=1, time=None, base_date_name='event__date_from', minutes=None, is_after=True))
+            else:
+                raise ValueError()
+
+        @pytest.fixture(params=["single_event", "subevents"])
+        def rdt_event_variants(self, request):
+            return request.param
+
+        @pytest.fixture
+        def rdt_events(self, rdt_event_variants, event):
+            if rdt_event_variants == "single_event":
+                event.date_from = self.REFERENCE_DT
+                event.save()
+            else:
+                event.has_subevents = True
+                event.subevents.create(
+                    name='1',
+                    date_from=self.REFERENCE_DT,
+                )
+                event.subevents.create(
+                    name='2',
+                    date_from=self.REFERENCE_DT + timedelta(days=1),
+                )
+                event.subevents.create(
+                    name='3',
+                    date_from=self.REFERENCE_DT + timedelta(days=2),
+                )
+            return event
+
+        @pytest.fixture
+        def rdt_item(self, rdt_events):
+            return rdt_events.items.create(
+                name='Ticket',
+                category=None, default_price=23,
+                admission=True
+            )
+
+        @pytest.fixture(params=["EARLIEST", "LATEST"])
+        def rdt_mode_variants(self, request):
+            return request.param
+
+        @pytest.fixture
+        def rdt_order(self, rdt_events):
+            o = Order.objects.create(
+                code='123456', event=rdt_events, email='dummy@dummy.test',
+                status=Order.STATUS_PENDING,
+                datetime=self.REFERENCE_DT + timedelta(hours=6),  # 6 hours offset mark orders
+                sales_channel=rdt_events.organizer.sales_channels.get(identifier="web"),
+                total=14, locale='en'
+            )
+            return o
+
+        @pytest.fixture
+        def rdt_order_positions(self, rdt_event_variants, rdt_events, rdt_item, rdt_order):
+            if rdt_event_variants == "single_event":
+                op = [OrderPosition.objects.create(
+                    order=rdt_order,
+                    item=rdt_item,
+                    variation=None,
+                    price=Decimal("14"),
+                )]
+            else:
+                op = []
+                for i in range(0, 3):
+                    op.append(OrderPosition.objects.create(
+                        subevent=rdt_events.subevents.all()[i],
+                        order=rdt_order,
+                        item=rdt_item,
+                        variation=None,
+                        price=Decimal("14"),
+                    ))
+            return op
+
+        @pytest.mark.django_db
+        def test_process_rule_resolve_date_field(
+                self,
+                rdt_reldate,
+                rdt_reldate_variants,
+                rdt_events,
+                rdt_event_variants,
+                rdt_mode_variants,
+                rdt_order,
+                rdt_order_positions
+        ):
+            with scope(organizer=rdt_events.organizer):
+                date = ProcessCancellationRule._resolve_date_field(rdt_reldate, rdt_order, rdt_mode_variants)
+                match rdt_reldate_variants:
+                    case "date":
+                        assert date == self.REFERENCE_DT
+                    case "datetime":
+                        assert date == self.REFERENCE_DT
+                    case "order":
+                        assert date == self.REFERENCE_DT + timedelta(days=1) + timedelta(hours=6)
+                    case "event":
+                        if rdt_event_variants == "single_event":
+                            assert date == self.REFERENCE_DT + timedelta(days=1)
+                        elif rdt_event_variants == "subevents":
+                            if rdt_mode_variants == "EARLIEST":
+                                assert date == self.REFERENCE_DT + timedelta(days=1)
+                            elif rdt_mode_variants == "LATEST":
+                                assert date == self.REFERENCE_DT + timedelta(days=1) + timedelta(days=2)
+                            else:
+                                raise ValueError("Variant not known")
+                        else:
+                            raise ValueError("Variant not known")
+                    case _:
+                        raise ValueError("Variant not known")
+
+        @pytest.mark.django_db
+        def test_position_rule_resolve_date_field(
+                self,
+                rdt_reldate,
+                rdt_reldate_variants,
+                rdt_events,
+                rdt_event_variants,
+                rdt_order,
+                rdt_order_positions
+        ):
+            with scope(organizer=rdt_events.organizer):
+                for pos in rdt_order_positions:
+                    date = PositionCancellationRule._resolve_date_field(rdt_reldate, rdt_order, pos)
+                    match rdt_reldate_variants:
+                        case "date":
+                            assert date == self.REFERENCE_DT
+                        case "datetime":
+                            assert date == self.REFERENCE_DT
+                        case "order":
+                            assert date == self.REFERENCE_DT + timedelta(days=1) + timedelta(hours=6)
+                        case "event":
+                            if rdt_event_variants == "single_event":
+                                assert date == self.REFERENCE_DT + timedelta(days=1)
+                            elif rdt_event_variants == "subevents":
+                                assert date == pos.subevent.date_from + timedelta(days=1)
+                            else:
+                                raise ValueError("Variant not known")
+                        case _:
+                            raise ValueError("Variant not known")
+
+    class TestPositionMatchesRule:
+        @pytest.fixture
+        def items(self, event):
+            return [event.items.create(
+                name='Product 1',
+                category=None, default_price=23,
+                admission=True
+            ), event.items.create(
+                name='Product 2',
+                category=None, default_price=23,
+                admission=True
+            )]
+
+        @pytest.fixture
+        def variations(self, event, items):
+            item = items[0]
+            return [
+                item.variations.create(
+                    value="Variation 1"
+                ),
+                item.variations.create(
+                    value="Variation 2"
+                ),
+
+            ]
+
+        @pytest.mark.django_db
+        @pytest.mark.parametrize(
+            ("item_idx", "variation_idx", "all_products", "limit_products", "limit_variations", "matches"),
+            [
+                (0, None, True, [], [], True),
+                (0, 0, True, [], [], True),
+                (0, 1, True, [], [], True),
+                (1, None, True, [], [], True),
+                (0, None, False, [], [], False),
+                (0, 0, False, [], [], False),
+                (0, 1, False, [], [], False),
+                (1, None, False, [], [], False),
+                (0, None, False, [0], [], True),
+                (1, None, False, [0], [], False),
+                (0, None, False, [1], [], False),
+                (0, 0, False, [0], [0], True),
+                (1, None, False, [0], [0], False),
+                (0, 1, False, [1], [], False),
+                (0, 0, False, [0], [0, 1], True),
+            ],
+            ids=[
+                "all_products::item-0",
+                "all_products::item-0-variation-0",
+                "all_products::item-0-variation-1",
+                "all_products::item-1",
+                "no-product::item-0",
+                "no-product::item-0-variation-0",
+                "no-product::item-0-variation-1",
+                "no-product::item-1",
+                "item-0::item-0",
+                "item-0::item-1",
+                "item-1::item-0",
+                "item-0-variation-0::item-0-variation-0",
+                "item-0-variation-0::item-1",
+                "item-1::item-0-variation-1",
+                "item-0-variation-0-variation-1::item-0-variation-0",
+
+            ]
+
+        )
+        def test_position_matches_rule(self, event, order, items, variations, item_idx,
+                                       variation_idx, all_products, limit_products, limit_variations,
+                                       matches):
+            with scope(organizer=event.organizer):
+                op = OrderPosition.objects.create(
+                    order=order,
+                    item=items[item_idx],
+                    variation=variations[variation_idx] if variation_idx is not None else None,
+                    price=Decimal("14"),
+                )
+                r = PositionCancellationRule.objects.create(event=event, all_products=all_products)
+                for lp in limit_products:
+                    r.limit_products.add(items[lp])
+                for lv in limit_variations:
+                    r.limit_variations.add(variations[lv])
+
+                rule = PositionCancellationRule.objects.with_rule_data().get(id=r.id)
+                with ensure_no_queries():
+                    res = rule._position_matches_rule(op)
+                if not matches:
+                    assert res is None
+                else:
+                    assert matches == res.cancellation_possible
+
+    class TestEvaluateCancellationMoment:
+        @pytest.fixture(params=["position", "process"])
+        def rule_type_variants(self, request):
+            return request.param
+
+        @pytest.mark.django_db
+        @pytest.mark.parametrize(
+            ('attr', "delta", "allowed"),
+            [
+                ('allowed_until', timedelta(hours=-1), True),
+                ('allowed_until', timedelta(hours=0), True),
+                ('allowed_until', timedelta(hours=+1), False),
+                ('except_after', timedelta(hours=-1), True),
+                ('except_after', timedelta(hours=0), True),
+                ('except_after', timedelta(hours=+1), False)
+
+            ]
+        )
+        def test_evaluate_cancellation_moment(self, event, order, order_position, rule_type_variants, attr, delta,
+                                              allowed):
+            reference_ts = datetime(2020, 10, 1, hour=0, minute=0, second=0, microsecond=0, tzinfo=UTC)
+
+            with scope(organizer=event.organizer):
+                if rule_type_variants == 'position':
+                    rule_object = PositionCancellationRule
+                    r = rule_object.objects.create(event=event, all_products=True)
+                elif rule_type_variants == "process":
+                    rule_object = ProcessCancellationRule
+                    r = rule_object.objects.create(event=event, all_products=True, fee_mode=FeeType.MINIMUM)
+                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)
+
+                with ensure_no_queries():
+                    if rule_type_variants == 'position':
+                        res = rule._evaluate_cancellation_moment(position=order_position, check_ts=reference_ts + delta)
+                    elif rule_type_variants == "process":
+                        res = rule._evaluate_cancellation_moment(order=order,
+                                                                 check_ts=reference_ts + delta)
+                    else:
+                        raise ValueError("Unknown cancellation rule type: {}".format(rule_type_variants))
+
+            assert len(res) == 2
+            for r in res:
+                if attr in r.id:
+                    assert r.cancellation_possible == allowed
+
+    class TestEvaluate:
+        # TODO add more elaborate test cases
+
+        @pytest.mark.django_db
+        def test_evaluate_simple_e2e(self, event, order, order_position):
+            reference_ts = datetime(2020, 10, 1, hour=0, minute=0, second=0, microsecond=0, tzinfo=UTC)
+
+            check_ts = reference_ts - timedelta(hours=1)
+
+            PositionCancellationRule.objects.create(event=event, all_products=True,
+                                                    fee_absolute_per_position=Decimal("10.00"),
+                                                    allowed_until=RelativeDateWrapper(reference_ts))
+            PositionCancellationRule.objects.create(event=event, all_products=True,
+                                                    fee_absolute_per_position=Decimal("10.00"),
+                                                    allowed_until=RelativeDateWrapper(reference_ts - timedelta(days=1)))
+            ProcessCancellationRule.objects.create(event=event, fee_cancellation_process=Decimal("10.00"),
+                                                   fee_mode=FeeType.ADDITIONAL,
+                                                   allowed_until=RelativeDateWrapper(reference_ts))
+            ProcessCancellationRule.objects.create(event=event, fee_cancellation_process=Decimal("10.00"),
+                                                   fee_mode=FeeType.ADDITIONAL,
+                                                   allowed_until=RelativeDateWrapper(reference_ts - timedelta(days=1)))
+
+            with scope(organizer=event.organizer):
+                cancellation = Cancellation.evaluate(event, order, keep=set(), check_ts=check_ts)
+
+            assert cancellation.possible == True
+
+            position_rule_results = cancellation.result.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
+
+            process_rule_results = cancellation.result.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

src/tests/helpers/test_database.py

@@ -0,0 +1,79 @@
+"""
+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 os
+
+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
+    os.environ.setdefault("ENSURE_NO_QUERIES_OVERRIDE", "true")
+
+    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