fix(core): correct CommandOutputs for outputs in subdirectories (#2798)

renku/core/management/workflow/plan_factory.py

@@ -410,6 +410,8 @@ def add_command_output(
         postfix: Optional[str] = None,
         encoding_format: List[str] = None,
         name: Optional[str] = None,
+        id: str = None,
+        mapped_to: Optional[MappedIOStream] = None,
     ):
         """Create a CommandOutput."""
         if self.no_output_detection and all(Path(default_value).resolve() != path for path, _ in self.explicit_outputs):
@@ -424,21 +426,23 @@ def add_command_output(
         ):
             create_folder = True
 
-        mapped_stream = self.get_stream_mapping_for_value(default_value)
+        mapped_stream = mapped_to or self.get_stream_mapping_for_value(default_value)
 
         if mapped_stream and position is None:
             position = (
                 max((p.position for p in chain(self.inputs, self.outputs, self.parameters) if p.position), default=0)
                 + 1
             )
 
+        id = id or CommandOutput.generate_id(
+            plan_id=self.plan_id,
+            position=position,
+            postfix=mapped_stream.stream_type if mapped_stream else postfix,
+        )
+
         self.outputs.append(
             CommandOutput(
-                id=CommandOutput.generate_id(
-                    plan_id=self.plan_id,
-                    position=position,
-                    postfix=mapped_stream.stream_type if mapped_stream else postfix,
-                ),
+                id=id,
                 default_value=default_value,
                 prefix=prefix,
                 position=position,
@@ -453,16 +457,15 @@ def add_command_output(
     def add_command_output_from_input(self, input: CommandInput, name):
         """Create a CommandOutput from an input."""
         self.inputs.remove(input)
-        self.outputs.append(
-            CommandOutput(
-                id=input.id.replace("/inputs/", "/outputs/"),
-                default_value=input.default_value,
-                prefix=input.prefix,
-                position=input.position,
-                mapped_to=input.mapped_to,
-                encoding_format=input.encoding_format,
-                name=name,
-            )
+
+        self.add_command_output(
+            id=input.id.replace("/inputs/", "/outputs/"),
+            default_value=input.default_value,
+            prefix=input.prefix,
+            position=input.position,
+            mapped_to=input.mapped_to,
+            encoding_format=input.encoding_format,
+            name=name,
         )
 
     def add_command_output_from_parameter(self, parameter: CommandParameter, name):

tests/cli/test_rerun.py

@@ -321,3 +321,20 @@ def test_rerun_multiple_paths_common_output(project, renku_cli, runner):
     assert "r2" not in result.output
     assert "r3" in result.output
     assert "r4" in result.output
+
+
+def test_rerun_output_in_subdirectory(split_runner, client):
+    """Test re-run when an output is in a sub-directory."""
+    output = client.path / "sub-dir" / "output"
+    write_and_commit_file(client.repository, output, "")
+
+    result = split_runner.invoke(cli, ["run", "bash", "-c", 'touch "$0" ; echo data > "$0"', output])
+
+    assert 0 == result.exit_code, format_result_exception(result)
+
+    write_and_commit_file(client.repository, output, "")
+
+    result = split_runner.invoke(cli, ["rerun", output])
+
+    assert 0 == result.exit_code, format_result_exception(result)
+    assert "data" in output.read_text()

tests/fixtures/repository.py

@@ -20,7 +20,7 @@
 import os
 import secrets
 import shutil
-from typing import Iterator
+from typing import Generator
 
 import pytest
 
@@ -124,7 +124,7 @@ def project(repository):
 
 
 @pytest.fixture
-def client(project, global_config_dir) -> Iterator[LocalClient]:
+def client(project, global_config_dir) -> Generator[LocalClient, None, None]:
     """Return a Renku repository."""
     from renku.core.models.enums import ConfigFilter