Adjust EventsAsyncService and fix tests

SDESK-7442

server/planning/events/events.py

@@ -49,9 +49,10 @@
 from apps.auth import get_user, get_user_id
 from apps.archive.common import get_auth, update_dates_for
 
-from planning.types import Event, PlanningRelatedEventLink, PLANNING_RELATED_EVENT_LINK_TYPE
+from planning.types import EmbeddedCoverageItem, Event, PlanningRelatedEventLink, PLANNING_RELATED_EVENT_LINK_TYPE
 from planning.types.event import EmbeddedPlanning
 from planning.common import (
+    TEMP_ID_PREFIX,
     UPDATE_SINGLE,
     UPDATE_FUTURE,
     get_max_recurrent_events,
@@ -79,7 +80,6 @@
 from .events_base_service import EventsBaseService
 from .events_schema import events_schema
 from .events_sync import sync_event_metadata_with_planning_items
-from .events_utils import get_events_embedded_planning
 
 logger = logging.getLogger(__name__)
 
@@ -95,6 +95,23 @@
 }
 
 
+# TODO-ASYNC: this method was migrated to events_utils and it uses pydantic models instead
+def get_events_embedded_planning(event: Event) -> List[EmbeddedPlanning]:
+    def get_coverage_id(coverage: EmbeddedCoverageItem) -> str:
+        if not coverage.get("coverage_id"):
+            coverage["coverage_id"] = TEMP_ID_PREFIX + "-" + generate_guid(type=GUID_NEWSML)
+        return coverage["coverage_id"]
+
+    return [
+        EmbeddedPlanning(
+            planning_id=planning.get("planning_id"),
+            update_method=planning.get("update_method") or "single",
+            coverages={get_coverage_id(coverage): coverage for coverage in planning.get("coverages") or []},
+        )
+        for planning in event.pop("embedded_planning", [])
+    ]
+
+
 def get_subject_str(subject: Dict[str, str]) -> str:
     return ":".join(
         [
@@ -199,6 +216,7 @@ def set_ingest_provider_sequence(item, provider):
     def on_create(self, docs):
         # events generated by recurring rules
         generated_events = []
+
         for event in docs:
             # generates an unique id
             if "guid" not in event:

server/planning/events/events_service.py

@@ -108,43 +108,65 @@ async def get_expired_items(
             # Yield the results for iteration by the callee
             yield items
 
+    def _extract_embedded_planning(
+        self, docs: list[EventResourceModel]
+    ) -> list[tuple[EventResourceModel, list[EmbeddedPlanning]]]:
+        """
+        Extracts out the ``embedded_planning`` of a given event Event
+        """
+
+        embedded_planning_lists: list[tuple[EventResourceModel, list[EmbeddedPlanning]]] = []
+        for event in docs:
+            embedded_planning = get_events_embedded_planning(event)
+            if len(embedded_planning):
+                embedded_planning_lists.append((event, embedded_planning))
+
+        return embedded_planning_lists
+
+    def _synchronise_associated_plannings(
+        self, embedded_planning_list: list[tuple[EventResourceModel, list[EmbeddedPlanning]]]
+    ):
+        """
+        Synchronise/process the given associated Planning item(s)
+        """
+        for event, embedded_planning in embedded_planning_list:
+            sync_event_metadata_with_planning_items(None, event.to_dict(), embedded_planning)
+
     async def create(self, docs: list[EventResourceModel]):
         """
         Extracts out the ``embedded_planning`` before saving the Event(s)
         And then uses them to synchronise/process the associated Planning item(s)
         """
 
         docs = await self._convert_dicts_to_model(docs)
-        print(docs)
+
+        embedded_planning_list = self._extract_embedded_planning(docs)
+        await self.prepare_events_data(docs)
         ids = await super().create(docs)
 
-        embedded_planning_lists: list[tuple[EventResourceModel, list[EmbeddedPlanning]]] = []
+        self._synchronise_associated_plannings(embedded_planning_list)
 
-        for event in docs:
-            embedded_planning = get_events_embedded_planning(event)
-            if len(embedded_planning):
-                embedded_planning_lists.append((event.to_dict(), embedded_planning))
+        return ids
 
-        if len(embedded_planning_lists):
-            for event, embedded_planning in embedded_planning_lists:
-                sync_event_metadata_with_planning_items(None, event, embedded_planning)
+    async def prepare_events_data(self, docs: list[EventResourceModel]) -> None:
+        """
+        Prepares basic attributes of events before creation. This method generates recurring events
+        if applicable, sets up planning schedules, and links events to planning items.
 
-        return ids
+        Args:
+            events (list[EventResourceModel]): A list of event models to prepare.
 
-    async def on_create(self, docs: list[EventResourceModel]) -> None:
-        # events generated by recurring rules
+        Returns:
+            None: Modifies the input list in-place.
+        """
         generated_events = []
         for event in docs:
-            # generates an unique id
             if not event.guid:
                 event.guid = generate_guid(type=GUID_NEWSML)
             event.id = event.guid
 
             if not event.language:
-                try:
-                    event.language = event.languages[0]
-                except IndexError:
-                    event.language = get_app_config("DEFAULT_LANGUAGE")
+                event.language = event.languages[0] if len(event.languages) > 0 else get_app_config("DEFAULT_LANGUAGE")
 
             # TODO-ASYNC: consider moving this into base service later
             event.original_creator = ObjectId(get_user_id()) or None
@@ -163,14 +185,16 @@ async def on_create(self, docs: list[EventResourceModel]) -> None:
             event.planning_schedule = self._create_planning_schedule(event)
             original_planning_item = event.planning_item
 
-            # validate event
             self.validate_event(event)
 
             # If _created_externally is true, generate_recurring_events is restricted.
             if event.dates and event.dates.recurring_rule and not event.dates.recurring_rule._created_externally:
                 recurring_events = self._generate_recurring_events(event)
                 generated_events.extend(recurring_events)
 
+                # remove the event that contains the recurring rule. We don't need it anymore
+                docs.remove(event)
+
                 # Set the current Event to the first Event in the new series
                 # This will make sure the ID of the Event can be used when
                 # using 'event' from here on, such as when linking to a Planning item
@@ -186,13 +210,11 @@ async def on_create(self, docs: list[EventResourceModel]) -> None:
 
             if original_planning_item:
                 await self._link_to_planning(event)
-                del event["_planning_item"]
+                event.planning_item = None
 
         if generated_events:
             docs.extend(generated_events)
 
-        await super().on_create(docs)
-
     async def on_created(self, docs: list[EventResourceModel]):
         """Send WebSocket Notifications for created Events
 
@@ -248,7 +270,6 @@ async def on_update(self, updates: dict[str, Any], original: EventResourceModel)
             return
 
         update_method = updates.pop("update_method", UpdateMethods.SINGLE)
-
         user = get_user()
         user_id = user.get(ID_FIELD) if user else None
 
@@ -263,7 +284,7 @@ async def on_update(self, updates: dict[str, Any], original: EventResourceModel)
             raise SuperdeskApiError.forbiddenError("The item was locked by another user")
 
         # If only the `recurring_rule` was provided, then fill in the rest from the original
-        # This can happen, for example, when converting a single Event to a series of Recurring Events
+        # this can happen, for example, when converting a single Event to a series of Recurring Events
         if list(updates.get("dates") or {}) == ["recurring_rule"]:
             new_dates = original.to_dict()["dates"]
             new_dates.update(updates["dates"])
@@ -470,10 +491,7 @@ async def _update_single_event(self, updates: dict[str, Any], original: EventRes
             get_resource_service("events_post").validate_item(merged.to_dict())
 
         # Determine if we're to convert this single event to a recurring of events
-        if (
-            original.lock_action == "convert_recurring"
-            and updates.get("dates", {}).get("recurring_rule", None) is not None
-        ):
+        if original.lock_action == "convert_recurring" and updates.get("dates", {}).get("recurring_rule") is not None:
             generated_events = await self._convert_to_recurring_events(updates, original)
 
             # if the original event was "posted" then post all the generated events
@@ -612,7 +630,10 @@ def mark_event_complete(self, updates: dict[str, Any], event: EventResourceModel
                 )
 
     async def _convert_to_recurring_events(self, updates: dict[str, Any], original: EventResourceModel):
-        """Convert a single event to a series of recurring events"""
+        """
+        Convert a single event to a series of recurring events and stores them into database.
+        This also triggers the `signals.events_created` signal.
+        """
 
         self._validate_convert_to_recurring(updates, original)
         updates["recurrence_id"] = original.id
@@ -652,8 +673,12 @@ async def _convert_to_recurring_events(self, updates: dict[str, Any], original:
             updates["_planning_schedule"] = [x.to_dict() for x in self._create_planning_schedule(updated_event)]
             remove_lock_information(item=updates)
 
-        # Create the new events and generate their history
-        await self.create(generated_events)
+        # create the new events
+        embedded_planning_list = self._extract_embedded_planning(generated_events)
+        await super().create(generated_events)
+        self._synchronise_associated_plannings(embedded_planning_list)
+
+        # signal's listener will generate these events' history
         await signals.events_created.send(generated_events)
 
         return generated_events

server/planning/events/events_sync/__init__.py

@@ -15,7 +15,7 @@
 
 from superdesk import get_resource_service
 
-from planning.utils import str_to_date
+from planning.utils import parse_date
 from planning.utils import get_related_planning_for_events
 from planning.content_profiles.utils import AllContentProfileData
 from planning.common import get_config_event_fields_to_sync_with_planning
@@ -43,13 +43,9 @@ def get_translated_fields(translations: List[StringFieldTranslation]) -> Dict[st
 # TODO-ASYNC: use resource models instead of typed dicts
 def sync_event_metadata_with_planning_items(
     original: Optional[Event],
-    updates: Event | EventResourceModel,
+    updates: Event,
     embedded_planning: list[EmbeddedPlanningDict] | list[EmbeddedPlanningModel],
 ):
-    # TODO-ASYNC: remove these checks after this is migrated
-    if isinstance(updates, EventResourceModel):
-        updates = cast(Event, updates.to_dict())
-
     embedded_planning = [
         cast(EmbeddedPlanningDict, obj.to_dict()) if isinstance(obj, EmbeddedPlanningModel) else obj
         for obj in embedded_planning
@@ -63,7 +59,7 @@ def sync_event_metadata_with_planning_items(
     event_updated.update(updates)
 
     if isinstance(event_updated["dates"]["start"], str):
-        event_updated["dates"]["start"] = str_to_date(event_updated["dates"]["start"])
+        event_updated["dates"]["start"] = parse_date(event_updated["dates"]["start"])
 
     if event_updated["dates"]["start"].tzinfo is None:
         event_updated["dates"]["start"] = event_updated["dates"]["start"].replace(tzinfo=pytz.utc)