fix(workflows): escape remaining unescaped Rich markup paths

Covers the last few review threads not yet addressed:
- Escape yaml.YAMLError text in the local workflow add install path
  (matches the already-escaped download/catalog paths).
- Escape the non---dev local directory fallback's "No workflow.yml
  found in <path>" message (the --dev branch already escaped it).
- Escape the redirected final_url in the --from non-HTTPS redirect
  error (IPv6 literals like http://[::1]/... are legal and contain
  brackets).
- Escape the "Downloaded workflow is invalid" exception message in
  _install_workflow_from_catalog, matching the sibling catalog-install
  exception handler a few lines above it.

Adds regression tests for each in TestWorkflowCliAlignment, following
the existing escaping-test pattern in this class.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
marcelsafin
2026-07-10 22:25:33 +02:00
parent 10f9770f0c
commit 91c71655c7
2 changed files with 111 additions and 4 deletions

View File

@@ -626,7 +626,7 @@ def workflow_add(
try:
definition = WorkflowDefinition.from_yaml(yaml_path)
except (ValueError, yaml.YAMLError) as exc:
console.print(f"[red]Error:[/red] Invalid workflow YAML: {exc}")
console.print(f"[red]Error:[/red] Invalid workflow YAML: {_escape_markup(str(exc))}")
raise typer.Exit(1)
# Non-string ids (e.g. unquoted ``id: 123`` or ``id: 0``) fall through
# to validate_workflow below, which reports a typed error instead of
@@ -739,7 +739,9 @@ def workflow_add(
# Redirect host is not an IP literal; keep loopback as determined above.
pass
if final_parsed.scheme != "https" and not (final_parsed.scheme == "http" and final_lb):
console.print(f"[red]Error:[/red] URL redirected to non-HTTPS: {final_url}")
console.print(
f"[red]Error:[/red] URL redirected to non-HTTPS: {_escape_markup(final_url)}"
)
raise typer.Exit(1)
with tempfile.NamedTemporaryFile(suffix=".yml", delete=False) as tmp:
tmp.write(resp.read())
@@ -770,7 +772,7 @@ def workflow_add(
elif source_path.is_dir():
wf_file = source_path / "workflow.yml"
if not wf_file.exists():
console.print(f"[red]Error:[/red] No workflow.yml found in {source}")
console.print(f"[red]Error:[/red] No workflow.yml found in {_escape_markup(source)}")
raise typer.Exit(1)
_validate_and_install_local(wf_file, str(source_path))
return
@@ -893,7 +895,7 @@ def _install_workflow_from_catalog(
except (ValueError, yaml.YAMLError) as exc:
import shutil
shutil.rmtree(workflow_dir, ignore_errors=True)
console.print(f"[red]Error:[/red] Downloaded workflow is invalid: {exc}")
console.print(f"[red]Error:[/red] Downloaded workflow is invalid: {_escape_markup(str(exc))}")
raise typer.Exit(1)
from .engine import validate_workflow

View File

@@ -7300,6 +7300,40 @@ steps:
assert result.exit_code != 0
assert "No workflow.yml found" in result.output
def test_add_local_dir_without_workflow_yml_errors(self, project_dir, monkeypatch):
"""Same as the --dev case, but for the plain local-path fallback (no --dev)."""
from typer.testing import CliRunner
from specify_cli import app
monkeypatch.chdir(project_dir)
empty = project_dir / "empty-src-[bracket]"
empty.mkdir()
runner = CliRunner()
result = runner.invoke(app, ["workflow", "add", str(empty)])
assert result.exit_code != 0
assert "No workflow.yml found" in result.output
assert "[bracket]" in result.output
def test_add_yaml_parse_error_escapes_rich_markup(self, project_dir, monkeypatch):
"""A YAML syntax error can quote the offending line verbatim; brackets in it must not be Rich markup."""
from unittest.mock import patch
from typer.testing import CliRunner
from specify_cli import app
from specify_cli.workflows.engine import WorkflowDefinition
monkeypatch.chdir(project_dir)
bad = project_dir / "bad.yml"
bad.write_text("workflow:\n id: wf\n", encoding="utf-8")
runner = CliRunner()
with patch.object(
WorkflowDefinition,
"from_yaml",
side_effect=ValueError('bad snippet: "New [Feature]"'),
):
result = runner.invoke(app, ["workflow", "add", str(bad)])
assert result.exit_code != 0
assert 'bad snippet: "New [Feature]"' in result.output
# -- add --from ----------------------------------------------------
class _FakeResponse:
@@ -7360,6 +7394,26 @@ steps:
assert "does not match" in result.output
assert not WorkflowRegistry(project_dir).is_installed("align-wf")
def test_add_from_url_non_https_redirect_escapes_rich_markup(self, project_dir, monkeypatch):
"""A redirect to a non-HTTPS IPv6 literal (legally bracketed) must not be parsed as Rich markup."""
from unittest.mock import patch
from typer.testing import CliRunner
from specify_cli import app
monkeypatch.chdir(project_dir)
redirected_url = "http://[2001:db8::1]/workflow.yml"
runner = CliRunner()
with patch(
"specify_cli.authentication.http.open_url",
side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(b"", redirected_url),
):
result = runner.invoke(
app,
["workflow", "add", "align-wf", "--from", "https://example.com/workflow.yml"],
)
assert result.exit_code != 0
assert redirected_url in result.output
def test_add_from_rejects_invalid_source_id_without_fetch(self, project_dir, monkeypatch):
"""--from with a non-workflow-id source (URL, path, uppercase) fails before any network fetch."""
from unittest.mock import patch
@@ -7514,6 +7568,57 @@ steps:
assert meta["version"] == "2.0.0"
assert "2.0.0" in (wf_dir / "workflow.yml").read_text(encoding="utf-8")
def test_update_downloaded_invalid_yaml_escapes_rich_markup(self, project_dir, monkeypatch):
"""A malformed downloaded workflow can quote the offending line verbatim; escape it before printing."""
from unittest.mock import patch
from typer.testing import CliRunner
from specify_cli import app
from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowRegistry
from specify_cli.workflows.engine import WorkflowDefinition
monkeypatch.chdir(project_dir)
registry = WorkflowRegistry(project_dir)
registry.add("align-wf", {
"name": "Align Workflow",
"version": "1.0.0",
"description": "CLI alignment test workflow",
"source": "catalog",
"catalog_name": "test-catalog",
"url": "https://example.com/workflow.yml",
})
wf_dir = project_dir / ".specify" / "workflows" / "align-wf"
wf_dir.mkdir(parents=True)
(wf_dir / "workflow.yml").write_text(
self.WORKFLOW_YAML.format(version="1.0.0"), encoding="utf-8"
)
monkeypatch.setattr(
WorkflowCatalog,
"get_workflow_info",
lambda self, wid: {
"id": wid,
"name": "Align Workflow",
"version": "2.0.0",
"url": "https://example.com/workflow.yml",
"_install_allowed": True,
"_catalog_name": "test-catalog",
},
)
runner = CliRunner()
with patch(
"specify_cli.authentication.http.open_url",
side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(b"", url),
), patch.object(
WorkflowDefinition,
"from_yaml",
side_effect=ValueError('bad snippet: "New [Feature]"'),
):
result = runner.invoke(app, ["workflow", "update"], input="y\n")
assert 'bad snippet: "New [Feature]"' in result.output
assert "Failed to update" in result.output
# The previously installed workflow must survive a failed update.
assert "1.0.0" in (wf_dir / "workflow.yml").read_text(encoding="utf-8")
def test_update_preserves_disabled_state(self, project_dir, monkeypatch):
from unittest.mock import patch
from typer.testing import CliRunner