make all fields in partial rerun merge events optional

Author: corva-alex-s

src/corva/handlers.py

@@ -576,6 +576,9 @@ def _merge_events(
     Only "scheduled" and "stream" type of apps can be processed here.
     If somehow any other type is passed - raise an exception
     """
+    if not isinstance(aws_event, list):     # if aws_event is not a list - there is nothing to merge, so do nothing.
+        return aws_event
+
     if data_transformation_type is RawScheduledEvent:
         # scheduled event
         if not isinstance(aws_event[0], dict):

src/corva/models/merge/merge.py

@@ -1,5 +1,6 @@
 from corva.models.base import CorvaBaseEvent
 from corva.models.merge.enums import EventType, RerunMode, SourceType
+from typing import Optional
 
 
 class PartialRerunMergeEvent(CorvaBaseEvent):
@@ -25,23 +26,23 @@ class PartialRerunMergeEvent(CorvaBaseEvent):
         run_until: run until
     """
 
-    event_type: EventType
-    partial_well_rerun_id: int
-    rerun_mode: RerunMode
-    start: int
-    end: int
-    asset_id: int
-    rerun_asset_id: int
-    app_stream_id: int
-    rerun_app_stream_id: int
-    version: int
-    app_id: int
-    app_key: str
-    app_connection_id: int
-    rerun_app_connection_id: int
-    source_type: SourceType
-    log_type: str
-    run_until: int
+    event_type: Optional[EventType]
+    partial_well_rerun_id: Optional[int]
+    rerun_mode: Optional[RerunMode]
+    start: Optional[int]
+    end: Optional[int]
+    asset_id: Optional[int]
+    rerun_asset_id: Optional[int]
+    app_stream_id: Optional[int]
+    rerun_app_stream_id: Optional[int]
+    version: Optional[int]
+    app_id: Optional[int]
+    app_key: Optional[str]
+    app_connection_id: Optional[int]
+    rerun_app_connection_id: Optional[int]
+    source_type: Optional[SourceType]
+    log_type: Optional[str]
+    run_until: Optional[int]
 
     class Config:
         extra = "allow"

src/corva/models/merge/raw.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 import pydantic
 
@@ -9,24 +9,24 @@
 
 
 class RawPartialMergeEventData(pydantic.BaseModel):
-    partial_well_rerun_id: int
-    rerun_mode: RerunMode
-    start: int
-    end: int
-    asset_id: int
-    rerun_asset_id: int
-    app_stream_id: int
-    rerun_app_stream_id: int
-    version: int = pydantic.Field(
+    partial_well_rerun_id: Optional[int]
+    rerun_mode: Optional[RerunMode]
+    start: Optional[int]
+    end: Optional[int]
+    asset_id: Optional[int]
+    rerun_asset_id: Optional[int]
+    app_stream_id: Optional[int]
+    rerun_app_stream_id: Optional[int]
+    version: Optional[int] = pydantic.Field(
         ..., le=1, ge=1
     )  # Currently handler supports only 1-st version of this event.
-    app_id: int
-    app_key: str
-    app_connection_id: int
-    rerun_app_connection_id: int
-    source_type: SourceType
-    log_type: str
-    run_until: int
+    app_id: Optional[int]
+    app_key: Optional[str]
+    app_connection_id: Optional[int]
+    rerun_app_connection_id: Optional[int]
+    source_type: Optional[SourceType]
+    log_type: Optional[str]
+    run_until: Optional[int]
 
     class Config:
         extra = "allow"

tests/unit/test_partial_rerun_merge_app.py

@@ -1,5 +1,5 @@
 """Module contains functional requirements for partial rerun merge event handler."""
-
+from copy import deepcopy
 from uuid import uuid4
 
 import pytest
@@ -33,7 +33,7 @@
 
 
 def test_merge_event_handler_called_from_stream_app_on_unexpected_event_type_raises_exc(
-    context,
+        context,
 ):
     """Partial merge app must raise Pydantic validation exception
     while processing event of unexpected type.
@@ -55,7 +55,7 @@ def partial_rerun_merge_app(event, api, asset_cache, rerun_asset_cache):
 
 
 def test_merge_event_handler_called_from_stream_app_returns_expected_cache_values(
-    context,
+        context,
 ):
     """Partial merge app must provide functional(working) cache objects
     for asset and rerun asset.
@@ -121,3 +121,39 @@ def partial_rerun_merge_app(event, api, asset_cache, rerun_asset_cache):
 
     partial_merge_app_response = scheduled_app(RAW_EVENT, context)[0]
     assert partial_merge_app_response == expected_response
+
+
+def test_event_parsing_not_failing_on_missing_field(context):
+    """When calling scheduled event with merge event handler defined,
+    partial merge handler should be called only.
+    """
+
+    @corva.scheduled(merge_events=True)
+    def scheduled_app(event, api, cache):
+        pytest.fail("Scheduled app was unexpectedly called!")
+
+    @corva.partial_rerun_merge
+    def partial_rerun_merge_app(event, api, asset_cache, rerun_asset_cache):
+        return True
+
+    event_with_missing_parameter = deepcopy(RAW_EVENT)
+    event_with_missing_parameter["data"]["end"] = None
+    partial_merge_app_response = scheduled_app(event_with_missing_parameter, context)[0]
+    assert partial_merge_app_response is True
+
+
+def test_merge_events_parameter_ignored_for_partial_rerun_merge(context):
+    """When calling scheduled event with merge event handler defined,
+    partial merge handler should be called only.
+    """
+
+    @corva.scheduled(merge_events=True)
+    def scheduled_app(event, api, cache):
+        pytest.fail("Scheduled app was unexpectedly called!")
+
+    @corva.partial_rerun_merge
+    def partial_rerun_merge_app(event, api, asset_cache, rerun_asset_cache):
+        return True
+
+    partial_merge_app_response = scheduled_app(RAW_EVENT, context)[0]
+    assert partial_merge_app_response is True