Subham-KRLX commented on code in PR #64051:
URL: https://github.com/apache/airflow/pull/64051#discussion_r3179693563
##########
providers/airbyte/tests/unit/airbyte/triggers/test_airbyte.py:
##########
@@ -166,60 +164,106 @@ async def test_airbyte_job_for_terminal_status_error(
"message": mock_message,
"job_id": self.JOB_ID,
}
- task = asyncio.create_task(trigger.run().__anext__())
- await asyncio.sleep(0.5)
- assert TriggerEvent(expected_result) == task.result()
- asyncio.get_event_loop().stop()
+
+ events = [e async for e in trigger.run()]
+ assert len(events) == 1
+ assert TriggerEvent(expected_result) == events[0]
@pytest.mark.asyncio
-
@mock.patch("airflow.providers.airbyte.triggers.airbyte.AirbyteSyncTrigger.is_still_running")
@mock.patch("airflow.providers.airbyte.hooks.airbyte.AirbyteHook.get_job_status")
- async def test_airbyte_job_exception(self, mock_get_job_status,
mocked_is_still_running):
+ async def test_airbyte_job_exception(self, mock_get_job_status):
"""Assert that run catch exception if Airbyte Sync job API throw
exception"""
- mocked_is_still_running.return_value = False
mock_get_job_status.side_effect = Exception("Test exception")
trigger = AirbyteSyncTrigger(
conn_id=self.CONN_ID,
poll_interval=self.POLL_INTERVAL,
end_time=self.END_TIME,
job_id=self.JOB_ID,
)
- task = [i async for i in trigger.run()]
- response = TriggerEvent(
+
+ events = [e async for e in trigger.run()]
+
+ expected_result = TriggerEvent(
{
"status": "error",
"message": "Test exception",
"job_id": self.JOB_ID,
}
)
- assert len(task) == 1
- assert response in task
+ assert len(events) == 1
+ assert expected_result in events
@pytest.mark.asyncio
-
@mock.patch("airflow.providers.airbyte.triggers.airbyte.AirbyteSyncTrigger.is_still_running")
@mock.patch("airflow.providers.airbyte.hooks.airbyte.AirbyteHook.get_job_status")
- async def test_airbyte_job_timeout(self, mock_get_job_status,
mocked_is_still_running):
+ async def test_airbyte_job_timeout(self, mock_get_job_status):
"""Assert that run timeout after end_time elapsed"""
- mocked_is_still_running.return_value = True
mock_get_job_status.side_effect = Exception("Test exception")
+
end_time = time.time()
trigger = AirbyteSyncTrigger(
conn_id=self.CONN_ID,
poll_interval=self.POLL_INTERVAL,
end_time=end_time,
job_id=self.JOB_ID,
)
- generator = trigger.run()
- actual = await generator.asend(None)
- expected = TriggerEvent(
+
+ events = [e async for e in trigger.run()]
+
+ expected_result = TriggerEvent(
{
"status": "error",
"message": f"Job run {self.JOB_ID} has not reached a terminal
status "
f"after {end_time} seconds.",
"job_id": self.JOB_ID,
}
)
- assert expected == actual
+
+ assert len(events) == 1
+ assert expected_result in events
+
+ @pytest.mark.asyncio
+ async def test_airbyte_job_run_execution_timeout(self):
+ """Assert that run timeout after execution_deadline has elapsed"""
+
+ execution_deadline = time.time() - 1
+
+ trigger = AirbyteSyncTrigger(
+ conn_id=self.CONN_ID,
+ poll_interval=self.POLL_INTERVAL,
+ end_time=time.time(),
+ execution_deadline=execution_deadline,
+ job_id=self.JOB_ID,
+ )
Review Comment:
Thanks, that correctly isolates the execution_deadline path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
