Added validation checks for import_tasks_from (ploomber#686)

* Added validation checks for import_tasks_from

* Removed unnecessary blank line in DAGSpecPartial

* Added tests for empty yaml and non-list yaml file

* Changed logic for handling empty YAML files and updated empty and non-list import_tasks_from tests

* Removed unnecessary blank lines

* Updated import_tasks_from tests for empty and non-list YAML file

src/ploomber/spec/dagspec.py

@@ -329,6 +329,18 @@ def _init(self, data, env, lazy_import, reload, parent_path,
                 imported = yaml.safe_load(
                     Path(self.data['meta']['import_tasks_from']).read_text())
 
+                if not imported:
+                    path = str(self.data['meta']['import_tasks_from'])
+                    raise ValueError('expected import_tasks_from file '
+                                     f'({path!r}) to return a list of tasks, '
+                                     f'got: {imported}')
+
+                if not isinstance(imported, list):
+                    raise TypeError(
+                        'Expected list when loading YAML file from '
+                        'import_tasks_from: file.yaml, '
+                        f'but got {type(imported)}')
+
                 if self.env is not None:
                     (imported,
                      tags_other) = expand_raw_dictionaries_and_extract_tags(

tests/spec/test_dagspec.py

@@ -1129,6 +1129,29 @@ def test_import_tasks_from(tmp_nbs):
     ]
 
 
+def test_import_tasks_from_empty_yaml_file(tmp_nbs):
+    Path('some_tasks.yaml').write_text('')
+
+    spec_d = yaml.safe_load(Path('pipeline.yaml').read_text())
+    spec_d['meta']['import_tasks_from'] = 'some_tasks.yaml'
+
+    with pytest.raises(ValueError) as excinfo:
+        DAGSpec(spec_d)
+    assert 'expected import_tasks_from' in str(excinfo.value)
+
+
+def test_import_tasks_from_non_list_yaml_file(tmp_nbs):
+    some_tasks = {'source': 'extra_task.py', 'product': 'extra.ipynb'}
+    Path('some_tasks.yaml').write_text(yaml.dump(some_tasks))
+
+    spec_d = yaml.safe_load(Path('pipeline.yaml').read_text())
+    spec_d['meta']['import_tasks_from'] = 'some_tasks.yaml'
+
+    with pytest.raises(TypeError) as excinfo:
+        DAGSpec(spec_d)
+    assert 'Expected list when loading YAML file' in str(excinfo.value)
+
+
 def test_import_tasks_from_does_not_resolve_dotted_paths(tmp_nbs):
     """
     Sources defined in a file used in "import_tasks_from" are resolved