**Description**

The operator is arguably unnecessarily limited to AI Platform’s standard images. The only change that is required to lift this constraint is making `package_uris` and `training_python_module` optional with default values `[]` and `None`, respectively. Then, using `master_config`, one can supply `imageUri` and run any image of choice.

**Use case / motivation**

This will open up for running arbitrary images on AI Platform.

**Are you willing to submit a PR?**

If the above sounds reasonable, I can open pull requests.
tests/providers/google/cloud/operators/test_mlengine.py
job_id='test_prediction',
                project_id='test-project',
                region=input_with_model['region'],
                data_format=input_with_model['dataFormat'],
                input_paths=input_with_model['inputPaths'],
                output_path=input_with_model['outputPath'],
                model_name=input_with_model['modelName'].split('/')[-1],
                dag=self.dag,
                task_id='test-prediction',
            )
            prediction_task.execute(None)
            mock_hook.assert_called_once_with(
                'google_cloud_default',
                None,
                impersonation_chain=None,
            )
            hook_instance.create_job.assert_called_once_with(
                'test-project', {'jobId': 'test_prediction', 'predictionInput': input_with_model}, ANY
            )
        self.assertEqual(http_error_code, context.exception.resp.status)
    @patch('airflow.providers.google.cloud.operators.mlengine.MLEngineHook')
    def test_failed_job_error(self, mock_hook):
        hook_instance = mock_hook.return_value
        hook_instance.create_job.return_value = {'state': 'FAILED', 'errorMessage': 'A failure message'}
        task_args = self.BATCH_PREDICTION_DEFAULT_ARGS.copy()
        task_args['uri'] = 'a uri'
        with self.assertRaises(RuntimeError) as context:
            MLEngineStartBatchPredictionJobOperator(**task_args).execute(None)
        self.assertEqual('A failure message', str(context.exception))
class TestMLEngineTrainingOperator(unittest.TestCase):