Linter&coverage

Author: corva-alex-s

src/corva/handlers.py

@@ -28,8 +28,7 @@
 from corva.models.merge.merge import PartialRerunMergeEvent
 from corva.models.merge.raw import RawPartialRerunMergeEvent
 from corva.models.scheduled.raw import RawScheduledEvent
-from corva.models.scheduled.scheduled import ScheduledEvent, ScheduledNaturalTimeEvent, ScheduledDataTimeEvent, \
-    ScheduledDepthEvent
+from corva.models.scheduled.scheduled import ScheduledEvent, ScheduledNaturalTimeEvent
 from corva.models.scheduled.scheduler_type import SchedulerType
 from corva.models.stream.raw import RawStreamEvent
 from corva.models.stream.stream import StreamEvent
@@ -132,14 +131,20 @@ def stream(
 
     Arguments:
         handler: logging handler to include in Corva logger.
-        merge_events: if True - merge all incoming events into one before passing them to func
+        merge_events: if True - merge all incoming events into one before
+        passing them to func
     """
 
     if func is None:
         return functools.partial(stream, handler=handler, merge_events=merge_events)
 
     @functools.wraps(func)
-    @functools.partial(base_handler, raw_event_type=RawStreamEvent, handler=handler, merge_events=merge_events)
+    @functools.partial(
+        base_handler,
+        raw_event_type=RawStreamEvent,
+        handler=handler,
+        merge_events=merge_events
+    )
     def wrapper(
         event: RawStreamEvent,
         api_key: str,
@@ -241,7 +246,12 @@ def scheduled(
         return functools.partial(scheduled, handler=handler, merge_events=merge_events)
 
     @functools.wraps(func)
-    @functools.partial(base_handler, raw_event_type=RawScheduledEvent, handler=handler, merge_events=merge_events)
+    @functools.partial(
+        base_handler,
+        raw_event_type=RawScheduledEvent,
+        handler=handler,
+        merge_events=merge_events
+    )
     def wrapper(
         event: RawScheduledEvent,
         api_key: str,
@@ -564,12 +574,18 @@ def _merge_events(aws_event: Any, data_transformation_type: RawBaseEvent) -> Any
         if not isinstance(aws_event[0], dict):
             aws_event: List[dict] = list(itertools.chain(*aws_event))
         is_depth = aws_event[0]["scheduler_type"] == SchedulerType.data_depth_milestone
-        event_start, event_end = ("top_depth", "bottom_depth") if is_depth else ("schedule_start", "schedule_end")
-        min_event_start, max_event_end = aws_event[0][event_start], aws_event[0].get(event_end)
+        event_start, event_end = ("top_depth", "bottom_depth")\
+            if is_depth else ("schedule_start", "schedule_end")
+        min_event_start, max_event_end = (aws_event[0][event_start],
+                                          aws_event[0].get(event_end))
         for event in aws_event[1:]:
             if event[event_start] < min_event_start:
                 min_event_start = event[event_start]
-            if max_event_end and event[event_end] > max_event_end:
+            if (
+                max_event_end
+                and event.get(event_end)
+                and event[event_end] > max_event_end
+            ):
                 max_event_end = event[event_end]
 
         aws_event[0][event_start] = min_event_start

tests/unit/test_scheduled_app.py

@@ -534,64 +534,39 @@ def scheduled_app(event, api, cache):
         scheduled_app(event, context)
 
 
-def test_merge_events_scheduled_event(context, requests_mock):
+@pytest.mark.parametrize(
+    "time_ranges, flat",
+    (
+        [((60, 120), (61, None), (62, 122)), True],
+        [((61, None), (60, None), (62, None)), False],
+    ),
+)
+def test_merge_events_scheduled_event(context, requests_mock, time_ranges, flat):
     @scheduled(merge_events=True)
     def scheduled_app(event, api, state):
         return event
 
-    event = [
-        [
-            RawScheduledDataTimeEvent(
-                asset_id=int(),
-                interval=60,
-                schedule=int(),
-                schedule_start=60,
-                schedule_end=120,
-                app_connection=int(),
-                app_stream=int(),
-                company=int(),
-                scheduler_type=SchedulerType.data_time,
-            ).dict(
-                by_alias=True,
-                exclude_unset=True,
-            ),
-            RawScheduledDataTimeEvent(
-                asset_id=int(),
-                interval=60,
-                schedule=int(),
-                schedule_start=61,
-                schedule_end=121,
-                app_connection=int(),
-                app_stream=int(),
-                company=int(),
-                scheduler_type=SchedulerType.data_time,
-            ).dict(
-                by_alias=True,
-                exclude_unset=True,
-            ),
+    event = []
+    for schedule_start, schedule_end in time_ranges:
+        event.append(
             RawScheduledDataTimeEvent(
                 asset_id=int(),
                 interval=60,
                 schedule=int(),
-                schedule_start=62,
-                schedule_end=122,
+                schedule_start=schedule_start,
+                schedule_end=schedule_end,
                 app_connection=int(),
                 app_stream=int(),
                 company=int(),
                 scheduler_type=SchedulerType.data_time,
-            ).dict(
-                by_alias=True,
-                exclude_unset=True,
-            )
-        ]
-    ]
-
-    # patch post request, that sets scheduled task as completed
-    # looks for url path like /scheduler/123/completed
-    post_mock = requests_mock.post(re.compile(r'/scheduler/\d+/completed'))
+            ).dict(by_alias=True, exclude_unset=True)
+        )
+    if not flat:
+        event = [event]
 
     result_event: ScheduledDataTimeEvent = scheduled_app(event, context)[0]
 
     assert result_event.start_time == 1
     assert result_event.end_time == 60
-    assert result_event.schedule_end == 122
+    max_schedule_run_value = time_ranges[-1][-1]
+    assert result_event.schedule_end == max_schedule_run_value

tests/unit/test_stream_app.py

@@ -621,8 +621,10 @@ def test_merge_events_stream_event(context):
     """
     merge_events parameter is merging records for "stream" apps
 
-    When 3 events with 2 records each are sent and @stream decorator has optional "merge_events" param set to True -
-    we're supposed to merge incoming events into one event with all records(6 in our case) combined
+    When 3 events with 2 records each are sent and @stream decorator
+    has optional "merge_events" param set to True - we're supposed
+    to merge incoming events into one event with all records(6 in our
+    case) combined
     """
     @stream(merge_events=True)
     def stream_app(event, api, cache):
@@ -634,12 +636,14 @@ def stream_app(event, api, cache):
                 RawTimeRecord(
                     collection=str(),
                     timestamp=1,
-                    **{"asset_id": 1, "company_id": 1},
+                    asset_id=1,
+                    company_id=1,
                 ),
                 RawTimeRecord(
                     collection=str(),
                     timestamp=2,
-                    **{"asset_id": 1, "company_id": 1},
+                    asset_id=1,
+                    company_id=1,
                 ),
             ],
             metadata=RawMetadata(
@@ -653,12 +657,14 @@ def stream_app(event, api, cache):
                 RawTimeRecord(
                     collection=str(),
                     timestamp=3,
-                    **{"asset_id": 1, "company_id": 1},
+                    asset_id=1,
+                    company_id=1,
                 ),
                 RawTimeRecord(
                     collection=str(),
                     timestamp=4,
-                    **{"asset_id": 1, "company_id": 1},
+                    asset_id=1,
+                    company_id=1,
                 )
             ],
             metadata=RawMetadata(
@@ -672,12 +678,14 @@ def stream_app(event, api, cache):
                 RawTimeRecord(
                     collection=str(),
                     timestamp=5,
-                    **{"asset_id": 1, "company_id": 1},
+                    asset_id=1,
+                    company_id=1,
                 ),
                 RawTimeRecord(
                     collection=str(),
                     timestamp=6,
-                    **{"asset_id": 1, "company_id": 1},
+                    asset_id=1,
+                    company_id=1,
                 )
             ],
             metadata=RawMetadata(