fix(core): fixes tests and duplicate objects in database, adds asciin…

…ema recordings

renku/cli/dataset.py

@@ -20,6 +20,10 @@
 Manipulating datasets
 ~~~~~~~~~~~~~~~~~~~~~
 
+.. image:: _static/asciicasts/dataset-create.delay.gif
+   :width: 600
+   :alt: Create a Dataset
+
 Creating an empty dataset inside a Renku project:
 
 .. code-block:: console
@@ -51,6 +55,10 @@
 
 Editing a dataset's metadata:
 
+.. image:: _static/asciicasts/dataset-edit.delay.gif
+   :width: 600
+   :alt: Editing a Dataset
+
 Use ``edit`` subcommand to change metadata of a dataset. You can edit the same
 set of metadata as the create command by passing the options described in the
 table above.
@@ -124,6 +132,9 @@
 Working with data
 ~~~~~~~~~~~~~~~~~
 
+.. image:: _static/asciicasts/dataset-add.delay.gif
+   :width: 600
+   :alt: Add data to a Dataset
 
 Adding data to the dataset:
 
@@ -298,6 +309,10 @@
 
 Importing data from an external provider:
 
+.. image:: _static/asciicasts/dataset-import.delay.gif
+   :width: 600
+   :alt: Import a Dataset
+
 .. code-block:: console
 
     $ renku dataset import 10.5281/zenodo.3352150

renku/cli/rerun.py

@@ -20,6 +20,10 @@
 Recreating files
 ~~~~~~~~~~~~~~~~
 
+.. image:: _static/asciicasts/rerun.delay.gif
+   :width: 600
+   :alt: Rerun workflow
+
 Assume you have run a step 2 that uses a stochastic algorithm, so each run
 will be slightly different. The goal is to regenerate output ``C`` several
 times to compare the output. In this situation it is not possible to simply

renku/cli/update.py

@@ -17,6 +17,10 @@
 # limitations under the License.
 r"""Update outdated files created by the "run" command.
 
+.. image:: _static/asciicasts/update.delay.gif
+   :width: 600
+   :alt: Update outdate files
+
 Recreating outdated files
 ~~~~~~~~~~~~~~~~~~~~~~~~~
 

renku/cli/workflow.py

@@ -43,6 +43,10 @@
 Listing Plans
 *************
 
+.. image:: _static/asciicasts/list_plans.delay.gif
+   :width: 600
+   :alt: List Plans
+
 .. code-block:: console
 
     $ renku workflow ls
@@ -60,6 +64,10 @@
 Showing Plan Details
 ********************
 
+.. image:: _static/asciicasts/show_plan.delay.gif
+   :width: 600
+   :alt: Show Plan
+
 You can see the details of a plan by using ``renku workflow show``:
 
 .. code-block:: console
@@ -86,6 +94,10 @@
 Executing Plans
 ***************
 
+.. image:: _static/asciicasts/execute_plan.delay.gif
+   :width: 600
+   :alt: Execute Plans
+
 Plans can be executed using ``renku workflow execute``. They can be run as-is
 or their parameters can be modified as needed. Renku has a plugin architecture
 to allow execution using various execution backends.
@@ -158,6 +170,9 @@
 
 Composing Plans into larger workflows
 *************************************
+.. image:: _static/asciicasts/compose_plan.delay.gif
+   :width: 600
+   :alt: Composing Plans
 
 For more complex workflows consisting of several steps, you can use the
 ``renku workflow compose`` command. This creates a new workflow that has
@@ -293,6 +308,10 @@
 Editing Plans
 *************
 
+.. image:: _static/asciicasts/edit_plan.delay.gif
+   :width: 600
+   :alt: Editing Plans
+
 Plans can be edited in some limited fashion, but we do not allow structural
 changes, as that might cause issues with the reproducibility and provenance of
 the project. If you want to do structural changes (e.g. adding/removing
@@ -344,6 +363,10 @@
 Visualizing Executions
 **********************
 
+.. image:: _static/asciicasts/visualize_runs.delay.gif
+   :width: 600
+   :alt: Visualizing Runs
+
 You can visualize past Runs made with renku using the ``renku workflow
 visualize`` command.
 This will show a directed graph of executions and how they are connected. This

renku/core/commands/workflow.py

@@ -512,15 +512,15 @@ def _nested_dict():
             set_param = reduce(lambda x, y: {y: x}, reversed(keys), value)
             override_params = always_merger.merge(override_params, set_param)
 
-    if override_params:
-        rv = ValueResolver.get(workflow, override_params)
-        workflow = rv.apply()
+    rv = ValueResolver.get(workflow, override_params)
 
-        if rv.missing_parameters:
-            communication.warn(
-                f'Could not resolve the following parameters in "{workflow.name}" workflow: '
-                f'{",".join(rv.missing_parameters)}'
-            )
+    workflow = rv.apply()
+
+    if rv.missing_parameters:
+        communication.warn(
+            f'Could not resolve the following parameters in "{workflow.name}" workflow: '
+            f'{",".join(rv.missing_parameters)}'
+        )
 
     if config:
         config = _safe_read_yaml(config)

renku/core/metadata/database.py

@@ -687,6 +687,9 @@ def __init__(self, database: Database):
         self._classes: Dict[str, type] = {}
         self._database = database
 
+        # a cache for normal (non-persistent objects with an id) to deduplicate them on load
+        self._normal_object_cache = {}
+
     def _get_class(self, type_name: str) -> type:
         cls = self._classes.get(type_name)
         if cls:
@@ -797,10 +800,16 @@ def _deserialize_helper(self, data, create=True):
                 data = self._deserialize_helper(data)
                 assert isinstance(data, dict)
 
+                if "id" in data and data["id"] in self._normal_object_cache:
+                    return self._normal_object_cache[data["id"]]
+
                 for name, value in data.items():
                     object.__setattr__(new_object, name, value)
 
                 if issubclass(cls, Immutable):
                     new_object = cls.make_instance(new_object)
 
+                if "id" in data and isinstance(data["id"], str) and data["id"].startswith("/"):
+                    self._normal_object_cache[data["id"]] = new_object
+
             return new_object

renku/core/models/workflow/composite_plan.py

@@ -211,6 +211,19 @@ def find_parameter(self, parameter: CommandParameterBase):
 
         return False
 
+    def get_parameter_by_id(self, parameter_id: str) -> CommandParameterBase:
+        """Get a parameter on this plan by id."""
+        mapping = next((p for p in self.mappings if parameter_id == p.id), None)
+
+        if mapping:
+            return mapping
+
+        for plan in self.plans:
+            parameter = plan.get_parameter_by_id(parameter_id)
+
+            if parameter:
+                return parameter
+
     def find_parameter_workflow(self, parameter: CommandParameterBase) -> Optional[Union["CompositePlan", Plan]]:
         """Return the workflow a parameter belongs to."""
         if parameter in self.mappings:

renku/core/models/workflow/plan.py

@@ -211,7 +211,11 @@ def resolve_direct_reference(self, reference: str) -> CommandParameterBase:
 
     def find_parameter(self, parameter: CommandParameterBase) -> bool:
         """Find if a parameter exists on this plan."""
-        return parameter in self.inputs + self.outputs + self.parameters
+        return any(parameter.id == p.id for p in self.inputs + self.outputs + self.parameters)
+
+    def get_parameter_by_id(self, parameter_id: str) -> CommandParameterBase:
+        """Get a parameter on this plan by id."""
+        return next((p for p in self.inputs + self.outputs + self.parameters if parameter_id == p.id), None)
 
     def find_parameter_workflow(self, parameter: CommandParameterBase) -> "Plan":
         """Return the workflow a parameter belongs to."""

tests/cli/fixtures/cli_repository.py

@@ -83,6 +83,7 @@ def repository(tmpdir):
             with GitConfigParser(get_config_path("global"), read_only=False) as global_config:
                 global_config.set_value("user", "name", "Renku @ SDSC")
                 global_config.set_value("user", "email", "renku@datascience.ch")
+                global_config.set_value("pull", "rebase", False)
 
             result = runner.invoke(cli, ["init", ".", "--template-id", "python-minimal"], "\n", catch_exceptions=False)
             assert 0 == result.exit_code, format_result_exception(result)

tests/cli/test_workflow.py

@@ -734,3 +734,54 @@ def _update_screen(data):
 
     child.expect(pexpect.EOF, timeout=2)
     assert not child.isalive()
+
+
+def test_workflow_compose_execute(runner, project, run_shell, client):
+    """Test renku workflow compose with execute."""
+    # Run a shell command with pipe.
+    output = run_shell('renku run --name run1 -- echo "a" > output1')
+
+    # Assert expected empty stdout.
+    assert b"" == output[0]
+    # Assert not allocated stderr.
+    assert output[1] is None
+
+    # Run a shell command with pipe.
+    output = run_shell("renku run --name run2 -- cp output1 output2")
+
+    # Assert expected empty stdout.
+    assert b"" == output[0]
+    # Assert not allocated stderr.
+    assert output[1] is None
+
+    # Run a shell command with pipe.
+    output = run_shell('renku run --name run3 -- echo "b" > output3')
+
+    # Assert expected empty stdout.
+    assert b"" == output[0]
+    # Assert not allocated stderr.
+    assert output[1] is None
+
+    # Run a shell command with pipe.
+    output = run_shell("renku run --name run4 -- cp output3 output4")
+
+    # Assert expected empty stdout.
+    assert b"" == output[0]
+    # Assert not allocated stderr.
+    assert output[1] is None
+
+    # we need to run in a subprocess to ensure the execute below uses a clean Database, to test against
+    # issues with cached parameters
+    output = run_shell("renku workflow compose --link run2.output-2=run4.input-1 composite_workflow1 run1 run2 run4")
+
+    # Assert not allocated stderr.
+    assert output[1] is None
+
+    assert "b\n" == Path("output4").read_text()
+
+    output = run_shell('renku workflow execute --set run1.parameter-1="xyz" composite_workflow1')
+
+    # Assert not allocated stderr.
+    assert output[1] is None
+
+    assert "xyz\n" == Path("output4").read_text()