SameerMesiah97 commented on code in PR #64051:
URL: https://github.com/apache/airflow/pull/64051#discussion_r3178600272
##########
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")
Review Comment:
`mock_get_job_status` is now needed under the new trigger structure as
job_status is now always checked before timeout evaluation. The `side_effect`
is now `RUNNING` to ensure that the timeout_evaluation codepath is reached.
--
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]
