https://github.com/apache/airflow/blob/fdd9b6f65b608c516b8a062b058972d9a45ec9e3/airflow/sensors/sql_sensor.py#L97


tests/sensors/test_sql_sensor.py
self.assertTrue(op.poke(None))
        mock_get_records.return_value = [['1']]
        self.assertFalse(op.poke(None))
    @mock.patch('airflow.sensors.sql_sensor.BaseHook')
    def test_sql_sensor_postgres_poke_failure(self, mock_hook):
        op = SqlSensor(
            task_id='sql_sensor_check', conn_id='postgres_default', sql="SELECT 1", failure=lambda x: x in [1]
        )
        mock_hook.get_connection('postgres_default').conn_type = "postgres"
        mock_get_records = mock_hook.get_connection.return_value.get_hook.return_value.get_records
        mock_get_records.return_value = []
        self.assertFalse(op.poke(None))
        mock_get_records.return_value = [[1]]
        self.assertRaises(AirflowException, op.poke, None)
    @mock.patch('airflow.sensors.sql_sensor.BaseHook')
    def test_sql_sensor_postgres_poke_failure_success(self, mock_hook):
        op = SqlSensor(
            task_id='sql_sensor_check',
            conn_id='postgres_default',
            sql="SELECT 1",
            failure=lambda x: x in [1],
            success=lambda x: x in [2],
        )
        mock_hook.get_connection('postgres_default').conn_type = "postgres"
        mock_get_records = mock_hook.get_connection.return_value.get_hook.return_value.get_records
        mock_get_records.return_value = []
        self.assertFalse(op.poke(None))
        mock_get_records.return_value = [[1]]
        self.assertRaises(AirflowException, op.poke, None)
        mock_get_records.return_value = [[2]]