Interpret pipeline libraries relative to spec (#368)

The library paths in a pipeline spec should be interpreted as relative to the spec's location, and not relative to the current working directory.

Author: anew

databricks_cli/pipelines/api.py

@@ -40,14 +40,14 @@ def __init__(self, api_client):
         self.client = DeltaPipelinesService(api_client)
         self.dbfs_client = DbfsApi(api_client)
 
-    def create(self, spec, allow_duplicate_names, headers=None):
-        data = self._upload_libraries_and_update_spec(spec)
+    def create(self, spec, spec_dir, allow_duplicate_names, headers=None):
+        data = self._upload_libraries_and_update_spec(spec, spec_dir)
         data['allow_duplicate_names'] = allow_duplicate_names
         return self.client.client.perform_query('POST', '/pipelines', data=data,
                                                 headers=headers)
 
-    def deploy(self, spec, allow_duplicate_names, headers=None):
-        data = self._upload_libraries_and_update_spec(spec)
+    def deploy(self, spec, spec_dir, allow_duplicate_names, headers=None):
+        data = self._upload_libraries_and_update_spec(spec, spec_dir)
         data['allow_duplicate_names'] = allow_duplicate_names
         pipeline_id = data['id']
         self.client.client.perform_query('PUT', '/pipelines/{}'.format(pipeline_id), data=data,
@@ -89,13 +89,13 @@ def run(self, pipeline_id, headers=None):
     def stop(self, pipeline_id, headers=None):
         self.client.stop(pipeline_id, headers)
 
-    def _upload_libraries_and_update_spec(self, spec):
+    def _upload_libraries_and_update_spec(self, spec, spec_dir):
         spec = copy.deepcopy(spec)
         lib_objects = LibraryObject.from_json(spec.get('libraries', []))
         local_lib_objects, external_lib_objects = self._identify_local_libraries(lib_objects)
 
-        spec['libraries'] = LibraryObject.to_json(external_lib_objects +
-                                                  self._upload_local_libraries(local_lib_objects))
+        spec['libraries'] = LibraryObject.to_json(
+            external_lib_objects + self._upload_local_libraries(spec_dir, local_lib_objects))
         return spec
 
     @staticmethod
@@ -127,14 +127,15 @@ def _identify_local_libraries(lib_objects):
                 external_lib_objects.append(lib_object)
         return local_lib_objects, external_lib_objects
 
-    def _upload_local_libraries(self, local_lib_objects):
-        remote_lib_objects = [LibraryObject(llo.lib_type, self._get_hashed_path(llo.path))
-                              for llo in local_lib_objects]
-
+    def _upload_local_libraries(self, spec_dir, local_lib_objects):
+        relative_local_lib_objects = [LibraryObject(llo.lib_type, os.path.join(spec_dir, llo.path))
+                                      for llo in local_lib_objects]
+        remote_lib_objects = [LibraryObject(rllo.lib_type, self._get_hashed_path(rllo.path))
+                              for rllo in relative_local_lib_objects]
         transformed_remote_lib_objects = [LibraryObject(rlo.lib_type, DbfsPath(rlo.path))
                                           for rlo in remote_lib_objects]
         upload_files = [llo_tuple for llo_tuple in
-                        zip(local_lib_objects, transformed_remote_lib_objects)
+                        zip(relative_local_lib_objects, transformed_remote_lib_objects)
                         if not self.dbfs_client.file_exists(llo_tuple[1].path)]
 
         for llo, rlo in upload_files:

databricks_cli/pipelines/cli.py

@@ -91,9 +91,10 @@ def deploy_cli(api_client, spec_arg, spec, allow_duplicate_names, pipeline_id):
         raise ValueError('The spec should be provided either by an option or argument')
     src = spec_arg if bool(spec_arg) else spec
     spec_obj = _read_spec(src)
+    spec_dir = os.path.dirname(src)
     if not pipeline_id and 'id' not in spec_obj:
         try:
-            response = PipelinesApi(api_client).create(spec_obj, allow_duplicate_names)
+            response = PipelinesApi(api_client).create(spec_obj, spec_dir, allow_duplicate_names)
         except requests.exceptions.HTTPError as e:
             _handle_duplicate_name_exception(spec_obj, e)
 
@@ -116,7 +117,7 @@ def deploy_cli(api_client, spec_arg, spec, allow_duplicate_names, pipeline_id):
         _validate_pipeline_id(spec_obj['id'])
 
         try:
-            PipelinesApi(api_client).deploy(spec_obj, allow_duplicate_names)
+            PipelinesApi(api_client).deploy(spec_obj, spec_dir, allow_duplicate_names)
         except requests.exceptions.HTTPError as e:
             _handle_duplicate_name_exception(spec_obj, e)
         click.echo("Successfully deployed pipeline: {}".format(
@@ -292,7 +293,9 @@ def _handle_duplicate_name_exception(spec, exception):
 
     if error_code == 'RESOURCE_CONFLICT':
         raise ValueError("Pipeline with name '{}' already exists. ".format(spec['name']) +
-                         "You can use the --allow-duplicate-names option to skip this check.")
+                         "If you are updating an existing pipeline, provide the pipeline " +
+                         "id using --pipeline-id. Otherwise, " +
+                         "you can use the --allow-duplicate-names option to skip this check. ")
     raise exception
 
 

tests/pipelines/test_api.py

@@ -103,18 +103,21 @@ def _test_library_uploads(pipelines_api, api_method, spec, put_file_mock, dbfs_p
     # set-up the test
     jar1 = tmpdir.join('jar1.jar').strpath
     jar2 = tmpdir.join('jar2.jar').strpath
-    jar3 = tmpdir.join('jar3.jar').strpath
+    jar3dir = 'some/relative/path'
+    jar3absdir = tmpdir.join(jar3dir).strpath
+    jar3relpath = os.path.join(jar3dir, 'jar3.jar')
+    jar3abspath = tmpdir.join(jar3relpath).strpath
     jar4 = tmpdir.join('jar4.jar').strpath
     wheel1 = tmpdir.join('wheel-name-conv.whl').strpath
-    jar3_relpath = os.path.relpath(jar3, os.getcwd())
     jar4_file_prefix = 'file:{}'.format(jar4)
     remote_path_456 = 'dbfs:/pipelines/code/51eac6b471a284d3341d8c0c63d0f1a286262a18.jar'
 
     with open(jar1, 'w') as f:
         f.write('123')
     with open(jar2, 'w') as f:
         f.write('456')
-    with open(jar3, 'w') as f:
+    os.makedirs(jar3absdir)
+    with open(jar3abspath, 'w') as f:
         f.write('456')
     with open(jar4, 'w') as f:
         f.write('456')
@@ -127,7 +130,7 @@ def _test_library_uploads(pipelines_api, api_method, spec, put_file_mock, dbfs_p
         {'unknown': '/foo/bar'},
         {'jar': jar1},
         {'jar': jar2},
-        {'jar': jar3_relpath},
+        {'jar': jar3relpath},
         {'jar': jar4_file_prefix},
         {'whl': wheel1},
     ]
@@ -152,12 +155,12 @@ def _test_library_uploads(pipelines_api, api_method, spec, put_file_mock, dbfs_p
     ]
     expected_data['allow_duplicate_names'] = allow_duplicate_names
 
-    api_method(spec, allow_duplicate_names)
+    api_method(spec, tmpdir.strpath, allow_duplicate_names)
     assert dbfs_path_validate.call_count == 5
     assert put_file_mock.call_count == 4
     assert put_file_mock.call_args_list[0][0][0] == jar2
     assert put_file_mock.call_args_list[0][0][1].absolute_path == remote_path_456
-    assert put_file_mock.call_args_list[1][0][0] == jar3_relpath
+    assert put_file_mock.call_args_list[1][0][0] == jar3abspath
     assert put_file_mock.call_args_list[2][0][0] == jar4
     assert put_file_mock.call_args_list[3][0][0] == wheel1
     client_mock = pipelines_api.client.client.perform_query
@@ -171,13 +174,13 @@ def test_create(pipelines_api):
     spec = copy.deepcopy(SPEC_WITHOUT_ID)
     spec['libraries'] = []
 
-    pipelines_api.create(spec, allow_duplicate_names=False)
+    pipelines_api.create(spec, spec_dir='.', allow_duplicate_names=False)
     data = copy.deepcopy(spec)
     data['allow_duplicate_names'] = False
     client_mock.assert_called_with("POST", "/pipelines", data=data, headers=None)
     assert client_mock.call_count == 1
 
-    pipelines_api.create(spec, allow_duplicate_names=True, headers=HEADERS)
+    pipelines_api.create(spec, spec_dir='.', allow_duplicate_names=True, headers=HEADERS)
     data = copy.deepcopy(spec)
     data['allow_duplicate_names'] = True
     client_mock.assert_called_with("POST", "/pipelines", data=data, headers=HEADERS)
@@ -190,13 +193,13 @@ def test_deploy(pipelines_api):
     spec = copy.deepcopy(SPEC)
     spec['libraries'] = []
 
-    pipelines_api.deploy(spec, allow_duplicate_names=False)
+    pipelines_api.deploy(spec, spec_dir='.', allow_duplicate_names=False)
     data = copy.deepcopy(spec)
     data['allow_duplicate_names'] = False
     client_mock.assert_called_with("PUT", "/pipelines/" + PIPELINE_ID, data=data, headers=None)
     assert client_mock.call_count == 1
 
-    pipelines_api.deploy(spec, allow_duplicate_names=True, headers=HEADERS)
+    pipelines_api.deploy(spec, spec_dir='.', allow_duplicate_names=True, headers=HEADERS)
     data = copy.deepcopy(spec)
     data['allow_duplicate_names'] = True
     client_mock.assert_called_with("PUT", "/pipelines/" + PIPELINE_ID, data=data, headers=HEADERS)

tests/pipelines/test_cli.py

@@ -273,19 +273,19 @@ def test_allow_duplicate_names_flag(pipelines_api_mock, tmpdir):
         f.write(DEPLOY_SPEC_NO_ID)
     runner = CliRunner()
     runner.invoke(cli.deploy_cli, [path])
-    assert pipelines_api_mock.create.call_args_list[0][0][1] is False
+    assert pipelines_api_mock.create.call_args_list[0][0][2] is False
 
     runner.invoke(cli.deploy_cli, [path, "--allow-duplicate-names"])
-    assert pipelines_api_mock.create.call_args_list[1][0][1] is True
+    assert pipelines_api_mock.create.call_args_list[1][0][2] is True
 
     with open(path, 'w') as f:
         f.write(DEPLOY_SPEC)
 
     runner.invoke(cli.deploy_cli, [path])
-    assert pipelines_api_mock.deploy.call_args_list[0][0][1] is False
+    assert pipelines_api_mock.deploy.call_args_list[0][0][2] is False
 
     runner.invoke(cli.deploy_cli, [path, "--allow-duplicate-names"])
-    assert pipelines_api_mock.deploy.call_args_list[1][0][1] is True
+    assert pipelines_api_mock.deploy.call_args_list[1][0][2] is True
 
 
 @provide_conf