Compatible blocks empty workflow up5561 (#252)

* add test_get_compatible_blocks_empty_workflow that returns the data blocks

* add workflow_mock_empty

* add block types to mock block definition

* return data blocks for empty workflow

* ignore types

* new coverage

* mute logger for under the hood get_block call

tests/fixtures.py

@@ -111,16 +111,19 @@
             "id": "4ed70368-d4e1-4462-bef6-14e768049471",
             "name": "tiling",
             "displayName": "Raster Tiling",
+            "type": "PROCESSING",
         },
         {
             "id": "c0d04ec3-98d7-4183-902f-5bcb2a176d89",
             "name": "sharpening",
             "displayName": "Sharpening Filter",
+            "type": "PROCESSING",
         },
         {
             "id": "c4cb8913-2ef3-4e82-a426-65ea8faacd9a",
             "name": "esa-s2-l2a-gtiff-visual",
             "displayName": "Sentinel-2 L2A Visual (GeoTIFF)",
+            "type": "DATA",
         },
     ],
     "error": {},
@@ -365,6 +368,50 @@ def project_live(auth_live):
     return project
 
 
+@pytest.fixture()
+def workflow_mock_empty(auth_mock, requests_mock):
+    """
+    Only to test error handling of functions that don't correctly work on workflows
+    without tasks (blocks). For the fully mocked workflow see workflow_mock fixture.
+    """
+    # info
+    url_workflow_info = (
+        f"{auth_mock._endpoint()}/projects/"
+        f"{auth_mock.project_id}/workflows/"
+        f"{WORKFLOW_ID}"
+    )
+    json_workflow_info = {
+        "data": {
+            "name": WORKFLOW_NAME,
+            "id": WORKFLOW_ID,
+            "description": PROJECT_DESCRIPTION,
+            "createdAt": "some_date",
+            "xyz": 789,
+        },
+        "error": {},
+    }
+    requests_mock.get(url=url_workflow_info, json=json_workflow_info)
+
+    workflow = Workflow(
+        auth=auth_mock,
+        workflow_id=WORKFLOW_ID,
+        project_id=auth_mock.project_id,
+    )
+
+    # get_workflow_tasks
+    json_empty_workflow_tasks = {
+        "error": "None",
+        "data": [],
+    }
+    url_workflow_tasks = (
+        f"{workflow.auth._endpoint()}/projects/{workflow.auth.project_id}/workflows/"
+        f"{workflow.workflow_id}/tasks"
+    )
+    requests_mock.get(url=url_workflow_tasks, json=json_empty_workflow_tasks)
+
+    return workflow
+
+
 @pytest.fixture()
 def workflow_mock(auth_mock, requests_mock):
     # info

tests/test_workflow.py

@@ -10,6 +10,7 @@
 from .fixtures import (
     auth_mock,
     auth_live,
+    workflow_mock_empty,
     workflow_mock,
     workflow_live,
     job_mock,
@@ -68,6 +69,13 @@ def test_get_compatible_blocks_live(workflow_live):
     assert "tiling" in list(compatible_blocks.keys())
 
 
+def test_get_compatible_blocks_empty_workflow_returns_data_blocks(workflow_mock_empty):
+    compatible_blocks = workflow_mock_empty.get_compatible_blocks()
+    assert isinstance(compatible_blocks, dict)
+    assert len(compatible_blocks) == 1
+    assert "esa-s2-l2a-gtiff-visual" in list(compatible_blocks.keys())
+
+
 def test_construct_full_workflow_tasks_dict_unkwown_block_raises(workflow_mock):
     input_tasks = ["some_block"]
     with pytest.raises(ValueError):

up42/workflow.py

@@ -91,7 +91,18 @@ def get_compatible_blocks(self) -> dict:
 
         Currently no data blocks can be attached to other data blocks.
         """
-        last_task = list(self.get_workflow_tasks(basic=True).keys())[-1]  # type: ignore
+        tasks: dict = self.get_workflow_tasks(basic=True)  # type: ignore
+        if not tasks:
+            logger.info("The workflow is empty, returning all data blocks.")
+
+            logging.getLogger("up42.tools").setLevel(logging.CRITICAL)
+            data_blocks = Tools(auth=self.auth).get_blocks(
+                block_type="data", basic=True
+            )
+            logging.getLogger("up42.tools").setLevel(logging.INFO)
+            return data_blocks  # type: ignore
+
+        last_task = list(tasks.keys())[-1]
         url = (
             f"{self.auth._endpoint()}/projects/{self.project_id}/workflows/{self.workflow_id}/"
             f"compatible-blocks?parentTaskName={last_task}"