From fba01d2a0787e074f1d7f76544539cbf5ea267f6 Mon Sep 17 00:00:00 2001 From: Alan Malta Rodrigues Date: Tue, 14 Jul 2026 11:31:13 +0200 Subject: [PATCH 1/4] refactor: use cwltool loader for standardized cwl doc handling --- src/dirac_cwl/job/executor/__main__.py | 200 ++++++++++--------------- 1 file changed, 80 insertions(+), 120 deletions(-) diff --git a/src/dirac_cwl/job/executor/__main__.py b/src/dirac_cwl/job/executor/__main__.py index eaa117e..6cfcc33 100644 --- a/src/dirac_cwl/job/executor/__main__.py +++ b/src/dirac_cwl/job/executor/__main__.py @@ -26,6 +26,11 @@ def _get_package_version(package: str) -> str: return "unknown" +def _strip_prefix(x: str) -> str: + """Utilitarian function to strip .# prefix from a string.""" + return x.replace(".#", "") + + # Create Typer app with context settings to allow extra arguments app = typer.Typer(context_settings={"allow_extra_args": True, "ignore_unknown_options": True}) @@ -52,12 +57,16 @@ def version_callback(value: bool): def print_workflow_visualization(workflow_path: Path): """Print a nice visualization of the workflow structure with graph representation.""" - import yaml + from cwl_utils.pack import pack + from cwl_utils.parser import load_document try: - with open(workflow_path, "r") as f: - cwl = yaml.safe_load(f) + cwl = load_document(pack(str(workflow_path)), baseuri=".") + except Exception as e: + console.print(f"[red]⚠ Could not load workflow:[/red] {e}\n") + raise typer.Exit(1) from e + try: console.print() console.print( Panel.fit( @@ -67,146 +76,97 @@ def print_workflow_visualization(workflow_path: Path): ) # Show basic info - cwl_version = cwl.get("cwlVersion", "Unknown") - doc = cwl.get("doc", cwl.get("label", "")) + cwl_version = getattr(cwl, "cwlVersion", "Unknown") + description = getattr(cwl, "doc", getattr(cwl, "label", "")) - info_table = Table(show_header=False, box=None, padding=(0, 2)) - info_table.add_column("Key", style="bold cyan") - info_table.add_column("Value") + console.print("[bold cyan]CWL Version:[/bold cyan]", cwl_version) + console.print("[bold cyan]Description:[/bold cyan]", description) - info_table.add_row("CWL Version:", cwl_version) - if doc: - info_table.add_row("Description:", doc) - - console.print(info_table) - console.print() - - # Show inputs (handle both dict and list formats) - inputs = cwl.get("inputs", {}) + # Show inputs + inputs = getattr(cwl, "inputs", []) if inputs: console.print("[bold green]📥 INPUTS:[/bold green]") - if isinstance(inputs, dict): - for name, spec in inputs.items(): - input_type = spec.get("type", "unknown") if isinstance(spec, dict) else spec - label = spec.get("label", "") if isinstance(spec, dict) else "" - label_str = f" [dim]({label})[/dim]" if label else "" - console.print(f" • [cyan]{name}[/cyan]: {input_type}{label_str}") - elif isinstance(inputs, list): - for inp in inputs: - if isinstance(inp, dict): - name = inp.get("id", "unknown") - input_type = inp.get("type", "unknown") - label = inp.get("label", inp.get("doc", "")) - label_str = f" [dim]({label})[/dim]" if label else "" - console.print(f" • [cyan]{name}[/cyan]: {input_type}{label_str}") + for inp_item in inputs: + inp_name = _strip_prefix(getattr(inp_item, "id", "unknown")) + inp_type = getattr(inp_item, "type_", "unknown") + inp_label = getattr(inp_item, "label", getattr(inp_item, "doc", "")) + label_str = f" [dim]({inp_label})[/dim]" if inp_label else "" + console.print(f" • [cyan]{inp_name}[/cyan]: {inp_type}{label_str}") console.print() - # Build and show graph representation (handle both dict and list formats) - steps = cwl.get("steps", {}) - outputs = cwl.get("outputs", {}) + # Build and show graph representation + steps = getattr(cwl, "steps", []) + num_steps = len(steps) + outputs = getattr(cwl, "outputs", []) if steps: console.print("[bold yellow]🔀 WORKFLOW GRAPH:[/bold yellow]") - console.print() - - # Build dependency graph (handle both dict and list formats) - if isinstance(steps, dict): - step_list = list(steps.items()) - elif isinstance(steps, list): - # Convert list format to (name, spec) tuples - step_list = [(s.get("id", f"step_{i}"), s) for i, s in enumerate(steps)] - else: - step_list = [] # Print graph representation - for i, (step_name, step_spec) in enumerate(step_list): - if isinstance(step_spec, dict): - is_last = i == len(step_list) - 1 - - # Print step box - step_prefix = "└──" if is_last else "├──" - step_label = step_spec.get("label", step_name) - console.print(f"{step_prefix} [bold cyan]{step_name}[/bold cyan] [dim]({step_label})[/dim]") - - # Indentation for details - detail_prefix = " " if is_last else "│ " - - # Show inputs with arrows (handle both dict and list formats) - step_in = step_spec.get("in", {}) - if step_in: - if isinstance(step_in, dict): - for in_name, in_source in step_in.items(): - source = ( - in_source - if isinstance(in_source, str) - else (in_source.get("source", "?") if isinstance(in_source, dict) else "?") - ) - console.print(f"{detail_prefix} [green]⬅[/green] {in_name} [dim]←[/dim] {source}") - elif isinstance(step_in, list): - for inp in step_in: - if isinstance(inp, dict): - in_name = inp.get("id", "?") - source = inp.get("source", "?") - console.print(f"{detail_prefix} [green]⬅[/green] {in_name} [dim]←[/dim] {source}") - - # Show outputs with arrows (handle both dict and list formats) - step_out = step_spec.get("out", []) - if step_out: - if isinstance(step_out, list): - for out in step_out: - out_name = out.get("id", out) if isinstance(out, dict) else out - console.print(f"{detail_prefix} [yellow]➡[/yellow] {out_name}") - - if not is_last: - console.print("│") + for i, step_spec in enumerate(steps): + is_last = i + 1 == num_steps + + # Print step box + step_prefix = "└──" if is_last else "├──" + step_name = _strip_prefix(getattr(step_spec, "id", "unknown")) + step_label = getattr(step_spec, "label", step_name) + step_label = step_label if step_label else step_name + console.print(f"{step_prefix} [bold cyan]{step_name}[/bold cyan] [dim]({step_label})[/dim]") + + # Indentation for details + detail_prefix = " " if is_last else "│ " + + # Show inputs with arrows + # note that there is no `in` attribute in this object, but there is an `in_` + for step_in in getattr(step_spec, "in_", []): + in_name = _strip_prefix(getattr(step_in, "id", "unknown")).replace(step_name + "/", "") + in_source = _strip_prefix(getattr(step_in, "source", "?")) + console.print(f"{detail_prefix} [green]⬅[/green] {in_name} [dim]←[/dim] {in_source}") + + # Show outputs with arrows + for step_out in getattr(step_spec, "out", []): + step_out_name = _strip_prefix(step_out).replace(step_name + "/", "") + console.print(f"{detail_prefix} [yellow]➡[/yellow] {step_out_name}") + + if not is_last: + console.print("│") console.print() # Show final outputs (handle both dict and list formats) if outputs: console.print("[bold magenta]📤 FINAL OUTPUTS:[/bold magenta]") - if isinstance(outputs, dict): - for name, spec in outputs.items(): - output_type = spec.get("type", "unknown") if isinstance(spec, dict) else spec - source = spec.get("outputSource", "") if isinstance(spec, dict) else "" - source_str = f" [dim]← {source}[/dim]" if source else "" - console.print(f" • [cyan]{name}[/cyan]: {output_type}{source_str}") - elif isinstance(outputs, list): - for out in outputs: - if isinstance(out, dict): - name = out.get("id", "unknown") - output_type = out.get("type", "unknown") - source = out.get("outputSource", "") - label = out.get("label", "") - label_str = f" [dim]({label})[/dim]" if label else "" - source_str = f" [dim]← {source}[/dim]" if source else "" - console.print(f" • [cyan]{name}[/cyan]: {output_type}{label_str}{source_str}") + for out_item in outputs: + out_name = _strip_prefix(getattr(out_item, "id", "unknown")) + # note that there is no `type` attribute in this object, but there is an `type_` + out_type = getattr(out_item, "type_", "unknown") + out_source = _strip_prefix(getattr(out_item, "outputSource", "?")) + console.print(f" • [cyan]{out_name}[/cyan]: {out_type} [dim]←[/dim] {out_source}") console.print() # Show hints - hints = cwl.get("hints", []) + hints = getattr(cwl, "hints", []) if hints: console.print("[bold blue]💡 HINTS:[/bold blue]") for hint in hints: - if isinstance(hint, dict): - hint_class = hint.get("class", "unknown") - console.print(f" • {hint_class}") - if hint_class == "dirac:Production": - # Use plugin system for display formatting - plugin_name = hint.get("input_dataset_plugin") - if plugin_name: - console.print(f" [dim]Plugin:[/dim] {plugin_name}") - try: - from dirac_cwl.production import get_registry - - plugin_cls = get_registry().get_plugin(plugin_name) - if plugin_cls: - config = hint.get("input_dataset_config", {}) - plugin_instance = plugin_cls() - for key, value in plugin_instance.format_hint_display(config): - console.print(f" [dim]{key}:[/dim] {value}") - except Exception: - pass # Silently ignore plugin display errors + hint_class = getattr(hint, "class", "unknown") + console.print(f" • {hint_class}") + if hint_class == "dirac:Production": + # Use plugin system for display formatting + plugin_name = hint.input_dataset_plugin + if plugin_name: + console.print(f" [dim]Plugin:[/dim] {plugin_name}") + try: + from dirac_cwl.production import get_registry + + plugin_cls = get_registry().get_plugin(plugin_name) + if plugin_cls: + config = hint.get("input_dataset_config", {}) + plugin_instance = plugin_cls() + for key, value in plugin_instance.format_hint_display(config): + console.print(f" [dim]{key}:[/dim] {value}") + except Exception: + pass # Silently ignore plugin display errors console.print() except Exception as e: From 5b17766462f6a002021093b9240d1e61b9e65d30 Mon Sep 17 00:00:00 2001 From: Alan Malta Rodrigues Date: Wed, 15 Jul 2026 09:40:25 +0200 Subject: [PATCH 2/4] test: unit tests for cwl job executor workflow visualization --- test/test_job_executor.py | 96 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 96 insertions(+) create mode 100644 test/test_job_executor.py diff --git a/test/test_job_executor.py b/test/test_job_executor.py new file mode 100644 index 0000000..9306d26 --- /dev/null +++ b/test/test_job_executor.py @@ -0,0 +1,96 @@ +"""Tests for the job executor CLI helpers.""" + +import re +from io import StringIO +from pathlib import Path +from unittest.mock import patch + +import pytest +import typer +from rich.console import Console + +from dirac_cwl.job.executor.__main__ import print_workflow_visualization + +REPO_ROOT = Path(__file__).resolve().parent.parent +HELLOWORLD_CWL = REPO_ROOT / "test/workflows/helloworld/description_basic.cwl" +CRYPTO_CWL = REPO_ROOT / "test/workflows/crypto/description.cwl" +MALFORMED_CLASS_CWL = REPO_ROOT / "test/workflows/malformed_description/description_malformed_class.cwl" +BAD_REFERENCE_CWL = REPO_ROOT / "test/workflows/bad_references/reference_doesnotexists.cwl" + + +def strip_ansi_codes(text: str) -> str: + """Remove ANSI color codes from text.""" + return re.sub(r"\x1b\[[0-9;]*m", "", text) + + +def capture_workflow_visualization(workflow_path: Path) -> str: + """Run print_workflow_visualization and return captured console output.""" + buffer = StringIO() + test_console = Console(file=buffer, width=120, force_terminal=True) + with patch("dirac_cwl.job.executor.__main__.console", test_console): + print_workflow_visualization(workflow_path) + return strip_ansi_codes(buffer.getvalue()) + + +class TestPrintWorkflowVisualization: + """Tests for print_workflow_visualization() function""" + + def test_command_line_tool_shows_metadata_only(self): + """A CommandLineTool gets a graceful inputs/outputs-only style view.""" + output = capture_workflow_visualization(HELLOWORLD_CWL) + + assert "Could not visualize workflow" not in output, output + assert "Workflow Visualization" in output + assert "description_basic.cwl" in output + assert "CWL Version: v1.2" in output + assert "WORKFLOW GRAPH" not in output + assert "INPUTS:" not in output + assert "FINAL OUTPUTS:" not in output + + def test_workflow_shows_inputs_steps_and_outputs(self): + """A workflow renders inputs, step wiring, and final outputs.""" + output = capture_workflow_visualization(CRYPTO_CWL) + + assert "Could not visualize workflow" not in output, output + assert "Workflow Visualization" in output + assert "description.cwl" in output + assert "CWL Version: v1.2" in output + assert "cryptographic transformations" in output + + assert "INPUTS:" in output + assert "input_string: string" in output + assert "shift_value: int" in output + + assert "WORKFLOW GRAPH:" in output + for step in ("caesar_step", "base64_step", "md5_step", "rot13_step"): + assert step in output + assert "input_string ← input_string" in output + assert "shift_value ← shift_value" in output + + assert "FINAL OUTPUTS:" in output + assert "caesar_output: File ← caesar_step/output" in output + assert "base64_output: File ← base64_step/output" in output + assert "md5_output: File ← md5_step/output" in output + assert "rot13_output: File ← rot13_step/output" in output + + @pytest.mark.parametrize( + "workflow_path", + [ + MALFORMED_CLASS_CWL, + BAD_REFERENCE_CWL, + REPO_ROOT / "test/workflows/does_not_exist.cwl", + ], + ) + def test_invalid_workflow_exits_with_error(self, workflow_path: Path): + """Invalid or missing workflows fail during load with a clear message.""" + expected_message = "Could not load workflow" + buffer = StringIO() + test_console = Console(file=buffer, width=120, force_terminal=True) + + with patch("dirac_cwl.job.executor.__main__.console", test_console): + with pytest.raises(typer.Exit) as exc_info: + print_workflow_visualization(workflow_path) + + assert exc_info.value.exit_code == 1 + output = strip_ansi_codes(buffer.getvalue()) + assert expected_message in output From 69ce84ad2f86e1b10bbd430899bc1233fea9f3e0 Mon Sep 17 00:00:00 2001 From: Alan Malta Rodrigues Date: Wed, 15 Jul 2026 09:43:04 +0200 Subject: [PATCH 3/4] test: add final period in method docstring --- test/test_job_executor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/test_job_executor.py b/test/test_job_executor.py index 9306d26..085fd8c 100644 --- a/test/test_job_executor.py +++ b/test/test_job_executor.py @@ -33,7 +33,7 @@ def capture_workflow_visualization(workflow_path: Path) -> str: class TestPrintWorkflowVisualization: - """Tests for print_workflow_visualization() function""" + """Tests for print_workflow_visualization() function.""" def test_command_line_tool_shows_metadata_only(self): """A CommandLineTool gets a graceful inputs/outputs-only style view.""" From 8211a48dc4875db06d413693797c9769d8a37855 Mon Sep 17 00:00:00 2001 From: Alan Malta Rodrigues Date: Thu, 23 Jul 2026 13:05:59 +0200 Subject: [PATCH 4/4] fix: apply suggestions from review --- src/dirac_cwl/job/executor/__main__.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/dirac_cwl/job/executor/__main__.py b/src/dirac_cwl/job/executor/__main__.py index 6cfcc33..6a9738f 100644 --- a/src/dirac_cwl/job/executor/__main__.py +++ b/src/dirac_cwl/job/executor/__main__.py @@ -76,7 +76,7 @@ def print_workflow_visualization(workflow_path: Path): ) # Show basic info - cwl_version = getattr(cwl, "cwlVersion", "Unknown") + cwl_version = cwl.cwlVersion description = getattr(cwl, "doc", getattr(cwl, "label", "")) console.print("[bold cyan]CWL Version:[/bold cyan]", cwl_version) @@ -133,7 +133,7 @@ def print_workflow_visualization(workflow_path: Path): console.print() - # Show final outputs (handle both dict and list formats) + # Show final outputs if outputs: console.print("[bold magenta]📤 FINAL OUTPUTS:[/bold magenta]") for out_item in outputs: