Fix pipelines request paths for delete, get, and reset commands (#274)

Previous change passed unit tests which made sure the correct call was made to the generated client, but didn't verify the generated client actually makes the right requests. Fixed service.py to send the right requests and updated unit tests to make sure the called paths are correct.

Also made the `databricks pipelines get` command print the server's response.

Author: mukulmurthy

databricks_cli/pipelines/api.py

@@ -48,16 +48,14 @@ def deploy(self, spec, headers=None):
 
         spec['libraries'] = LibraryObject.to_json(external_lib_objects +
                                                   self._upload_local_libraries(local_lib_objects))
-        self.client.client.perform_query('PUT',
-                                         '/pipelines/{}'.format(spec['id']),
-                                         data=spec,
-                                         headers=headers)
+        pipeline_id = spec['id']
+        self.client.deploy_spec(pipeline_id, spec, headers)
 
     def delete(self, pipeline_id, headers=None):
         self.client.delete(pipeline_id, headers)
 
     def get(self, pipeline_id, headers=None):
-        self.client.get(pipeline_id, headers)
+        return self.client.get(pipeline_id, headers)
 
     def reset(self, pipeline_id, headers=None):
         self.client.reset(pipeline_id, headers)

databricks_cli/pipelines/cli.py

@@ -27,7 +27,7 @@
 import click
 
 from databricks_cli.click_types import PipelineSpecClickType, PipelineIdClickType
-from databricks_cli.utils import eat_exceptions, CONTEXT_SETTINGS
+from databricks_cli.utils import eat_exceptions, CONTEXT_SETTINGS, pretty_format
 from databricks_cli.version import print_version_callback, version
 from databricks_cli.pipelines.api import PipelinesApi
 from databricks_cli.configure.config import provide_api_client, profile_option, debug_option
@@ -120,7 +120,7 @@ def get_cli(api_client, spec_arg, spec, pipeline_id):
     databricks pipelines get --pipeline-id 1234
     """
     pipeline_id = _get_pipeline_id(spec_arg=spec_arg, spec=spec, pipeline_id=pipeline_id)
-    PipelinesApi(api_client).get(pipeline_id)
+    click.echo(pretty_format(PipelinesApi(api_client).get(pipeline_id)))
 
 
 @click.command(context_settings=CONTEXT_SETTINGS,

databricks_cli/sdk/service.py

@@ -825,21 +825,24 @@ def deploy(self, pipeline_id=None, id=None, name=None, storage=None, configurati
                 raise TypeError('Expected databricks.Filters() or dict for field filters')
         return self.client.perform_query('PUT', '/pipelines/{pipeline_id}', data=_data, headers=headers)
 
+    def deploy_spec(self, pipeline_id=None, spec={}, headers=None):
+        return self.client.perform_query('PUT', '/pipelines/{}'.format(pipeline_id), data=spec, headers=headers)
+
     def delete(self, pipeline_id=None, headers=None):
         _data = {}
         if pipeline_id is not None:
             _data['pipeline_id'] = pipeline_id
-        return self.client.perform_query('DELETE', '/pipelines/{pipeline_id}', data=_data, headers=headers)
+        return self.client.perform_query('DELETE', '/pipelines/{}'.format(pipeline_id), data=_data, headers=headers)
 
     def get(self, pipeline_id=None, headers=None):
         _data = {}
         if pipeline_id is not None:
             _data['pipeline_id'] = pipeline_id
-        return self.client.perform_query('GET', '/pipelines/{pipeline_id}', data=_data, headers=headers)
+        return self.client.perform_query('GET', '/pipelines/{}'.format(pipeline_id), data={}, headers=headers)
 
     def reset(self, pipeline_id=None, headers=None):
         _data = {}
         if pipeline_id is not None:
             _data['pipeline_id'] = pipeline_id
-        return self.client.perform_query('POST', '/pipelines/{pipeline_id}/reset', data=_data, headers=headers)
+        return self.client.perform_query('POST', '/pipelines/{}/reset'.format(pipeline_id), data=_data, headers=headers)
 

tests/pipelines/test_api.py

@@ -37,18 +37,25 @@
 PIPELINE_ID = '123456'
 SPEC = {
     'id': PIPELINE_ID,
-    'name': 'test_pipeline'
+    'name': 'test_pipeline',
+    'storage': 'dbfs:/path'
 }
-HEADERS = 'dummy_headers'
+ID_ONLY_SPEC = {
+    'pipeline_id': PIPELINE_ID,
+}
+HEADERS = {'dummy_header': 'dummy_value'}
 
 
 @pytest.fixture()
 def pipelines_api():
-    with mock.patch('databricks_cli.pipelines.api.DeltaPipelinesService') \
-            as DeltaPipelinesServiceMock:
-        DeltaPipelinesServiceMock.return_value = mock.MagicMock()
-        _pipelines_api = api.PipelinesApi(None)
-        yield _pipelines_api
+    client_mock = mock.MagicMock()
+
+    def server_response(*args, **kwargs):
+        if args[0] == 'GET':
+            return {'pipeline_id': PIPELINE_ID, 'state': 'RUNNING'}
+    client_mock.perform_query = mock.MagicMock(side_effect=server_response)
+    _pipelines_api = api.PipelinesApi(client_mock)
+    yield _pipelines_api
 
 
 def file_exists_stub(_, dbfs_path):
@@ -75,7 +82,6 @@ def test_deploy(put_file_mock, dbfs_path_validate, pipelines_api, tmpdir):
     A test local file which has '456' written to it is not present in Dbfs and therefore must be.
     uploaded to dbfs.
     """
-    deploy_mock = pipelines_api.client.client.perform_query
     # set-up the test
     jar1 = tmpdir.join('jar1.jar').strpath
     jar2 = tmpdir.join('jar2.jar').strpath
@@ -122,39 +128,42 @@ def test_deploy(put_file_mock, dbfs_path_validate, pipelines_api, tmpdir):
     assert put_file_mock.call_args_list[1][0][0] == jar3_relpath
     assert put_file_mock.call_args_list[2][0][0] == jar4
     assert put_file_mock.call_args_list[3][0][0] == wheel1
-    deploy_mock.assert_called_with('PUT', '/pipelines/{}'.format(PIPELINE_ID),
+    client_mock = pipelines_api.client.client.perform_query
+    client_mock.assert_called_with('PUT', '/pipelines/' + PIPELINE_ID,
                                    data=expected_spec, headers=None)
 
     pipelines_api.deploy(spec, HEADERS)
-    deploy_mock.assert_called_with('PUT', '/pipelines/{}'.format(PIPELINE_ID),
+    client_mock.assert_called_with('PUT', '/pipelines/' + PIPELINE_ID,
                                    data=expected_spec, headers=HEADERS)
+    assert client_mock.call_count == 2
 
 
 def test_delete(pipelines_api):
     pipelines_api.delete(PIPELINE_ID)
-    delete_mock = pipelines_api.client.delete
-    assert delete_mock.call_count == 1
-    assert delete_mock.call_args[0][0] == PIPELINE_ID
-    assert delete_mock.call_args[0][1] is None
+    client_mock = pipelines_api.client.client.perform_query
+    assert client_mock.call_count == 1
+    client_mock.assert_called_with('DELETE', '/pipelines/' + PIPELINE_ID,
+                                   data=ID_ONLY_SPEC, headers=None)
 
     pipelines_api.delete(PIPELINE_ID, HEADERS)
-    assert delete_mock.call_args[0][0] == PIPELINE_ID
-    assert delete_mock.call_args[0][1] == HEADERS
+    client_mock.assert_called_with('DELETE', '/pipelines/' + PIPELINE_ID,
+                                   data=ID_ONLY_SPEC, headers=HEADERS)
 
 
 def test_get(pipelines_api):
-    pipelines_api.get(PIPELINE_ID)
-    get_mock = pipelines_api.client.get
-    assert get_mock.call_count == 1
-    assert get_mock.call_args[0][0] == PIPELINE_ID
+    response = pipelines_api.get(PIPELINE_ID)
+    client_mock = pipelines_api.client.client.perform_query
+    assert client_mock.call_count == 1
+    client_mock.assert_called_with('GET', '/pipelines/' + PIPELINE_ID, data={}, headers=None)
+    assert(response['pipeline_id'] == PIPELINE_ID and response['state'] == 'RUNNING')
 
 
 def test_reset(pipelines_api):
     pipelines_api.reset(PIPELINE_ID)
-    reset_mock = pipelines_api.client.reset
-    assert reset_mock.call_count == 1
-    assert reset_mock.call_args[0][0] == PIPELINE_ID
-    assert reset_mock.call_args[0][1] is None
+    client_mock = pipelines_api.client.client.perform_query
+    assert client_mock.call_count == 1
+    client_mock.assert_called_with('POST', '/pipelines/{}/reset'.format(PIPELINE_ID),
+                                   data=ID_ONLY_SPEC, headers=None)
 
 
 def test_partition_local_remote(pipelines_api):