[IMP] slot: tests, availability check

Author: tde-banana-odoo

addons/event/models/event_event.py

@@ -572,87 +572,6 @@ def _check_slots_dates(self):
                     slots="\n".join(f"- {slot.name}" for slot in slots_outside_event_bounds)
                 ))
 
-    # @api.constrains('seats_max', 'seats_limited', 'registration_ids')
-    # def _check_seats_availability(self, minimal_availability=0):
-    #     sold_out_events = []
-    #     for event in self:
-    #         if event.seats_limited and event.seats_max and event.seats_available < minimal_availability:
-    #             sold_out_events.append(_(
-    #                 '- "%(event_name)s": Missing %(nb_too_many)i seats.',
-    #                 event_name=event.name,
-    #                 nb_too_many=minimal_availability - event.seats_available,
-    #             ))
-    #     if sold_out_events:
-    #         raise ValidationError(_('There are not enough seats available for:')
-    #                               + '\n%s\n' % '\n'.join(sold_out_events))
-
-    def _verify_seats_availability(self, slot_id=False, ticket_id=False, minimal_availability=0):
-        """ Check the number of seats available for the event.
-        :slot_id: for the specified slot if specified.
-        :ticket_id: for the specified ticket if specified.
-        :minimal_availability: A minimal availability can be specified to ensure there is
-                               at least this number of seats left. Useful in the sale flow.
-        Raises:
-            ValidationError: If the event / slot / ticket / slot ticket do not have enough seats available.
-        """
-        sold_out_records = []
-        slot_tickets_nb_registrations = {
-            (slot.id, ticket.id): count
-            for (slot, ticket, count) in self.env['event.registration']._read_group(
-                domain=[('event_id', 'in', self.ids), ('state', 'in', ['open', 'done']), ('active', '=', True)],
-                groupby=['event_slot_id', 'event_ticket_id'],
-                aggregates=['__count']
-            )
-        }
-        for event in self:
-            slot = False
-            ticket = False
-            if slot_id:
-                slot = self.env['event.slot'].browse(slot_id)
-            if ticket_id:
-                ticket = self.env['event.event.ticket'].browse(ticket_id)
-            # Event / slot seats_max
-            if event.seats_limited and event.seats_max:
-                # TODO: ne fonctionne pas parce que ça verifie les valeurs avant le write et on ne sait pas
-                # cb de registrations sont impactée, on pourrait vouloir write sur 6 registrations
-                if slot and slot.seats_available < minimal_availability:
-                    sold_out_records.append(_(
-                        '- the slot "%(slot_name)s" (%(event_name)s): Missing %(nb_too_many)i seats.',
-                        slot_name=slot.name,
-                        event_name=event.name,
-                        nb_too_many=minimal_availability - slot.seats_available,
-                    ))
-                # TODO: same
-                elif event.seats_available < minimal_availability:
-                    sold_out_records.append(_(
-                        '- the event "%(event_name)s": Missing %(nb_too_many)i seats.',
-                        event_name=event.name,
-                        nb_too_many=minimal_availability - event.seats_available,
-                    ))
-            # Ticket seats_max
-            if ticket and ticket.seats_max:
-                if slot:
-                    # TODO: same
-                    slot_ticket_seats_available = ticket.seats_max - slot_tickets_nb_registrations.get((slot.id, ticket.id), 0)
-                    if slot_ticket_seats_available <= minimal_availability:
-                        sold_out_records.append(_(
-                            '- the slot ticket "%(slot_ticket_name)s" (%(event_name)s): Missing %(nb_too_many)i seats.',
-                            slot_ticket_name=f'{ticket.name} - {slot.name}',
-                            event_name=event.name,
-                            nb_too_many=minimal_availability - slot_ticket_seats_available,
-                        ))
-                # TODO: same
-                elif ticket.seats_available <= minimal_availability:
-                    sold_out_records.append(_(
-                        '- the ticket "%(ticket_name)s" (%(event_name)s): Missing %(nb_too_many)i seats.',
-                        ticket_name=ticket.name,
-                        event_name=event.name,
-                        nb_too_many=minimal_availability - ticket.seats_available,
-                    ))
-        if sold_out_records:
-            raise ValidationError(_('There are not enough seats available for:\n%(sold_out_records)s\n',
-                                    sold_out_records='\n'.join(sold_out_records)))
-
     @api.constrains('date_begin', 'date_end')
     def _check_closing_date(self):
         for event in self:
@@ -721,6 +640,84 @@ def _set_tz_context(self):
         self.ensure_one()
         return self.with_context(tz=self.date_tz or 'UTC')
 
+    def _get_seats_availability(self, slot_tickets):
+        """ Get availabilities for given combinations of slot / ticket. Returns
+        a list following input order. None denotes no limit. """
+        self.ensure_one()
+        if not (all(len(item) == 2 for item in slot_tickets)):
+            raise ValueError('Input should be a list of tuples containing slot, ticket')
+
+        slot_tickets_nb_registrations = {
+            (slot.id, ticket.id): count
+            for (slot, ticket, count) in self.env['event.registration']._read_group(
+                domain=[('event_id', 'in', self.ids), ('state', 'in', ['open', 'done']), ('active', '=', True)],
+                groupby=['event_slot_id', 'event_ticket_id'],
+                aggregates=['__count']
+            )
+        }
+
+        availabilities = []
+        for slot, ticket in slot_tickets:
+            available = None
+            # event is constrained: max stands for either each slot, either global (no slots)
+            if self.seats_max:
+                if slot:
+                    available = slot.seats_available
+                else:
+                    available = self.seats_available
+            # ticket is constrained: max standard for either each slot / ticket, either global (no slots)
+            if available != 0 and ticket and ticket.seats_max:
+                if slot:
+                    ticket_available = ticket.seats_max - slot_tickets_nb_registrations.get((slot.id, ticket.id), 0)
+                else:
+                    ticket_available = ticket.seats_available
+                available = ticket_available if available == None else min(available, ticket_available)
+            availabilities.append(available)
+        return availabilities
+
+    def _verify_seats_availability(self, slot_tickets):
+        """ Check event seats availability, for combinations of slot / ticket.
+
+        :slot_tickets: a list of tuples(slot, ticket, count). SLot and ticket
+          are optional, depending on event configuration. If count is 0
+          it is a simple check current values do not overflow limit. If count
+          is given, it serves as a check there are enough remaining seats.
+
+        Raises:
+            ValidationError: if the event / slot / ticket do not have enough
+            available seats
+        """
+        self.ensure_one()
+        if not (all(len(item) == 3 for item in slot_tickets)):
+            raise ValueError('Input should be a list of tuples containing slot, ticket, count')
+
+        sold_out = []
+        availabilities = self._get_seats_availability([(item[0], item[1]) for item in slot_tickets])
+        for (slot, ticket, count), available in zip(slot_tickets, availabilities, strict=True):
+            if available is None:  # unconstrained
+                continue
+            if available < count:
+                if slot and ticket:
+                    name = f'{ticket.name} - {slot.name}'
+                elif slot:
+                    name = slot.name
+                elif ticket:
+                    name = ticket.name
+                else:
+                    name = self.name
+                sold_out.append((name, count - available))
+
+        if sold_out:
+            info = []  # note: somehow using list comprehension make translate.py crash in default lang
+            for item in sold_out:
+                info.append(_('%(slot_name)s: missing %(count)s seats', slot_name=item[0], count=item[1]))
+            raise ValidationError(
+                _('There are not enough seats available for %(event_name)s:\n%(sold_out_info)s',
+                  event_name=self.name,
+                  sold_out_info='\n'.join(info),
+                )
+            )
+
     # ------------------------------------------------------------
     # ACTIONS
     # ------------------------------------------------------------

addons/event/models/event_registration.py

@@ -3,6 +3,8 @@
 import logging
 import os
 
+from collections import defaultdict
+
 from odoo import _, api, fields, models, SUPERUSER_ID
 from odoo.addons.event.tools.esc_label_tools import print_event_attendees, setup_printer, layout_96x82
 from odoo.tools import email_normalize, email_normalize_all, formataddr
@@ -95,11 +97,16 @@ def _get_random_barcode(self):
         'Barcode should be unique',
     )
 
-    # @api.constrains('state', 'event_id', 'event_ticket_id')
-    # def _check_seats_availability(self):
-    #     registrations_confirmed = self.filtered(lambda registration: registration.state in ('open', 'done'))
-    #     registrations_confirmed.event_id._check_seats_availability()
-    #     registrations_confirmed.event_ticket_id._check_seats_availability()
+    @api.constrains('active', 'state', 'event_id', 'event_slot_id', 'event_ticket_id')
+    def _check_seats_availability(self):
+        tocheck = self.filtered(lambda registration: registration.state in ('open', 'done') and registration.active)
+        for event, registrations in tocheck.grouped('event_id').items():
+            classified = defaultdict(lambda: self.env['event.registration'])
+            for reg in registrations:
+                classified[(reg.event_slot_id, reg.event_ticket_id)] += reg
+            event._verify_seats_availability(
+                [(key[0], key[1], 0) for key, regs in classified.items()],
+            )
 
     def default_get(self, fields):
         ret_vals = super().default_get(fields)
@@ -243,8 +250,6 @@ def create(self, vals_list):
         all_partner_ids = set(values['partner_id'] for values in vals_list if values.get('partner_id'))
         all_event_ids = set(values['event_id'] for values in vals_list if values.get('event_id'))
         for values in vals_list:
-            event = self.env['event.event'].browse(values['event_id'])
-            event._verify_seats_availability(slot_id=values.get('event_slot_id'), ticket_id=values.get('event_ticket_id'))
             if not values.get('phone'):
                 continue
 
@@ -257,49 +262,26 @@ def create(self, vals_list):
                 related_country = self.env.company.country_id
             values['phone'] = self._phone_format(number=values['phone'], country=related_country) or values['phone']
 
-        registrations = super(EventRegistration, self).create(vals_list)
+        registrations = super().create(vals_list)
         registrations._update_mail_schedulers()
         return registrations
 
     def write(self, vals):
-        if any(field in ['active', 'state', 'event_id', 'event_slot_id', 'event_ticket_id'] for field in vals):
-            self._verify_seats_availability()
         confirming = vals.get('state') in {'open', 'done'}
         to_confirm = (self.filtered(lambda registration: registration.state in {'draft', 'cancel'})
                       if confirming else None)
-        ret = super(EventRegistration, self).write(vals)
+        ret = super().write(vals)
         if confirming:
             to_confirm._update_mail_schedulers()
 
         return ret
 
-    def _verify_seats_availability(self):
-        """ Check event, slot and ticket seats availability for the registrations.
-        Raise Validation Error if not enough seats available.
-        """
-        regs_by_event = self._read_group([], groupby=['event_id'], aggregates=["id:recordset"])
-        for (event, regs) in regs_by_event:
-            grouped_regs = regs._read_group(
-                domain=[],
-                groupby=['event_slot_id', 'event_ticket_id']
-            )
-            for (slot, ticket) in grouped_regs:
-                event._verify_seats_availability(slot_id=slot.id, ticket_id=ticket.id)
-
     def _compute_display_name(self):
         """ Custom display_name in case a registration is nott linked to an attendee
         """
         for registration in self:
             registration.display_name = registration.name or f"#{registration.id}"
 
-    def action_unarchive(self):
-        res = super().action_unarchive()
-        # Necessary triggers as changing registration states cannot be used as triggers for the
-        # Event(Ticket) models constraints.
-        if unarchived := self.filtered(self._active_name):
-            unarchived._verify_seats_availability()
-        return res
-
     # ------------------------------------------------------------
     # ACTIONS / BUSINESS
     # ------------------------------------------------------------

addons/event/models/event_ticket.py

@@ -117,21 +117,6 @@ def _constrains_dates_coherency(self):
                 raise UserError(_('The stop date cannot be earlier than the start date. '
                                   'Please check ticket %(ticket_name)s', ticket_name=ticket.name))
 
-    # @api.constrains('registration_ids', 'seats_max')
-    # def _check_seats_availability(self, minimal_availability=0):
-    #     sold_out_tickets = []
-    #     for ticket in self:
-    #         if ticket.seats_max and ticket.seats_available < minimal_availability:
-    #             sold_out_tickets.append(_(
-    #                 '- the ticket "%(ticket_name)s" (%(event_name)s): Missing %(nb_too_many)i seats.',
-    #                 ticket_name=ticket.name,
-    #                 event_name=ticket.event_id.name,
-    #                 nb_too_many=minimal_availability - ticket.seats_available,
-    #             ))
-    #     if sold_out_tickets:
-    #         raise ValidationError(_('There are not enough seats available for:')
-    #                               + '\n%s\n' % '\n'.join(sold_out_tickets))
-
     @api.depends('seats_max', 'seats_available')
     @api.depends_context('name_with_seats_availability')
     def _compute_display_name(self):

addons/event/tests/__init__.py

@@ -3,4 +3,5 @@
 
 from . import test_event_internals
 from . import test_event_mail_schedule
+from . import test_event_slot
 from . import test_mailing

addons/event/tests/common.py

@@ -1,5 +1,6 @@
-# -*- coding: utf-8 -*-
-# Part of Odoo. See LICENSE file for full copyright and licensing details.
+from contextlib import contextmanager
+from freezegun import freeze_time
+from unittest.mock import patch
 
 from odoo import fields
 from odoo.addons.mail.tests.common import mail_new_test_user
@@ -157,6 +158,20 @@ def _create_registrations(cls, event, reg_count):
         } for idx in range(0, reg_count)])
         return registrations
 
+    @classmethod
+    def _create_registrations_for_slot_and_ticket(cls, event, slot, ticket, count, **add_values):
+        return cls.env['event.registration'].create([
+            dict(
+                    {
+                    'email': f'{slot.id if slot else "NoSlot"}.{ticket.id if ticket else "NoTicket"}@test.example.com',
+                    'event_id': event.id,
+                    'event_slot_id': slot.id if slot else False,
+                    'event_ticket_id': ticket.id if ticket else False,
+                    'name': f'{slot.id if slot else "NoSlot"}.{ticket.id if ticket else "NoTicket"}',
+                }, **add_values
+            ) for idx in range(0, count)
+        ])
+
     @classmethod
     def _setup_test_reports(cls):
         cls.test_report_view = cls.env["ir.ui.view"].create({
@@ -215,3 +230,12 @@ def assertSchedulerCronTriggers(self, capture, call_at_list):
         for record, call_at in zip(capture.records, call_at_list):
             self.assertEqual(record.call_at, call_at.replace(microsecond=0))
             self.assertEqual(record.cron_id, self.env.ref('event.event_mail_scheduler'))
+
+    @contextmanager
+    def mock_datetime_and_now(self, mock_dt):
+        """ Used when synchronization date (using env.cr.now()) is important
+        in addition to standard datetime mocks. Used mainly to detect sync
+        issues. """
+        with freeze_time(mock_dt), \
+             patch.object(self.env.cr, 'now', lambda: mock_dt):
+            yield