This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new abc3a20f080 Escape server and user text in airflowctl command messages
(#76265)
abc3a20f080 is described below
commit abc3a20f080a06dbd81b55fea028c22179cd6380
Author: Rishabh <[email protected]>
AuthorDate: Fri Nov 6 22:16:49 2026 +0530
Escape server and user text in airflowctl command messages (#76265)
* Escape server and user text in airflowctl command messages
Several airflowctl commands interpolated server error text, Dag run ids
and logical dates into Rich markup unescaped. A value containing a
tag-like fragment such as "[/x]" made rich.print raise MarkupError, so
the user saw a traceback instead of the actual error.
Escape the interpolated text in the dags, tasks, connections, auth and
config commands and in the Dag run lookup helpers.
* Escape variables import output and cover the escaped airflowctl messages
variables import printed the bulk result unescaped, and variable keys
can contain "[" and "/", so a key such as "a[/x]" crashed the success
message with MarkupError after the import had already been applied.
Escape it, and add tests for the task, connections, auth and config
messages escaped earlier in this PR so removing the escapes fails them.
* Escape Dag run ids in the clear prompt and cover remaining escapes
The server-provided dag_run_id was printed as Rich markup before the
confirmation prompt, so ids such as "[/x]" raised MarkupError and
"run[abc]" was listed as "run". Escape it.
Add tests so removing any of the existing escapes (next-execution error,
tasks states-for-dag-run, logical-date not found, Dag pause error, run id
not found) fails a test, and give the mocked clients added by the PR a
Client spec.
Claude-Session: https://claude.ai/code/session_01BVx6g9eNzpyiubnaPgUz9o
---------
Co-authored-by: Shahar Epstein <[email protected]>
---
.../src/airflowctl/ctl/commands/auth_command.py | 5 +-
.../src/airflowctl/ctl/commands/config_command.py | 3 +-
.../airflowctl/ctl/commands/connection_command.py | 5 +-
.../src/airflowctl/ctl/commands/dag_command.py | 10 +++-
.../src/airflowctl/ctl/commands/task_command.py | 11 ++--
airflow-ctl/src/airflowctl/ctl/utils/dag_run.py | 9 ++-
.../airflow_ctl/ctl/commands/test_auth_command.py | 52 +++++++++++++++-
.../ctl/commands/test_config_command.py | 14 ++++-
.../ctl/commands/test_connections_command.py | 25 ++++++++
.../airflow_ctl/ctl/commands/test_dag_command.py | 69 +++++++++++++++++++++-
.../airflow_ctl/ctl/commands/test_task_command.py | 48 +++++++++++++++
.../tests/airflow_ctl/ctl/utils/test_dag_run.py | 29 +++++++++
12 files changed, 262 insertions(+), 18 deletions(-)
diff --git a/airflow-ctl/src/airflowctl/ctl/commands/auth_command.py
b/airflow-ctl/src/airflowctl/ctl/commands/auth_command.py
index ee2a8512de7..6440000fce0 100644
--- a/airflow-ctl/src/airflowctl/ctl/commands/auth_command.py
+++ b/airflow-ctl/src/airflowctl/ctl/commands/auth_command.py
@@ -27,6 +27,7 @@ import sys
import keyring
import rich
from keyring.errors import NoKeyringError
+from rich.markup import escape
from airflowctl.api.client import (
NEW_API_CLIENT,
@@ -93,7 +94,7 @@ def login(args, api_client=NEW_API_CLIENT) -> None:
rich.print(success_message)
return
except Exception as e:
- rich.print(f"[red]Login failed: {e}")
+ rich.print(f"[red]Login failed: {escape(str(e))}")
sys.exit(1)
# Token-based login
@@ -156,7 +157,7 @@ def get_token(args, api_client=NEW_API_CLIENT) -> None:
)
print(login_response.access_token)
except Exception as e:
- rich.print(f"[red]Token generation failed: {e}[/red]", file=sys.stderr)
+ rich.print(f"[red]Token generation failed: {escape(str(e))}[/red]",
file=sys.stderr)
sys.exit(1)
diff --git a/airflow-ctl/src/airflowctl/ctl/commands/config_command.py
b/airflow-ctl/src/airflowctl/ctl/commands/config_command.py
index 30d77d1b298..eeb6e66aa5f 100644
--- a/airflow-ctl/src/airflowctl/ctl/commands/config_command.py
+++ b/airflow-ctl/src/airflowctl/ctl/commands/config_command.py
@@ -23,6 +23,7 @@ from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, NamedTuple
import rich
+from rich.markup import escape
from airflowctl.api.client import NEW_API_CLIENT, ClientKind,
provide_api_client
@@ -838,5 +839,5 @@ def lint(args, api_client=NEW_API_CLIENT) -> None:
rich.print("[green]No issues found in your airflow.cfg. It is
ready for Airflow 3![/green]")
except Exception as e:
- rich.print(f"[red]Lint configs failed: {e}")
+ rich.print(f"[red]Lint configs failed: {escape(str(e))}")
sys.exit(1)
diff --git a/airflow-ctl/src/airflowctl/ctl/commands/connection_command.py
b/airflow-ctl/src/airflowctl/ctl/commands/connection_command.py
index f99bd81cc7b..040f4e0960c 100644
--- a/airflow-ctl/src/airflowctl/ctl/commands/connection_command.py
+++ b/airflow-ctl/src/airflowctl/ctl/commands/connection_command.py
@@ -23,6 +23,7 @@ from typing import Any
import rich
from pydantic import ValidationError
+from rich.markup import escape
from airflowctl.api.client import NEW_API_CLIENT, ClientKind,
provide_api_client
from airflowctl.api.datamodels.generated import (
@@ -91,9 +92,9 @@ def import_(args, api_client=NEW_API_CLIENT) -> None:
)
response =
api_client.connections.bulk(BulkBodyConnectionBody(actions=[connection_create_action]))
if response.create.errors:
- rich.print(f"[red]Failed to import connections:
{response.create.errors}[/red]")
+ rich.print(f"[red]Failed to import connections:
{escape(str(response.create.errors))}[/red]")
raise SystemExit(1)
rich.print(f"[green]Successfully imported {response.create.success}
connection(s)[/green]")
except Exception as e:
- rich.print(f"[red]Failed to import connections: {e}[/red]")
+ rich.print(f"[red]Failed to import connections:
{escape(str(e))}[/red]")
raise SystemExit(1)
diff --git a/airflow-ctl/src/airflowctl/ctl/commands/dag_command.py
b/airflow-ctl/src/airflowctl/ctl/commands/dag_command.py
index 04c0e42cd10..812b3eec357 100644
--- a/airflow-ctl/src/airflowctl/ctl/commands/dag_command.py
+++ b/airflow-ctl/src/airflowctl/ctl/commands/dag_command.py
@@ -23,6 +23,7 @@ import sys
from typing import Literal
import rich
+from rich.markup import escape
from rich.text import Text
from airflowctl.api.client import (
@@ -55,7 +56,10 @@ def update_dag_state(
try:
response = api_client.dags.update(dag_id=dag_id, dag_body=dag_body)
except ServerResponseError as e:
- rich.print(f"[red]Error while trying to {operation} Dag {dag_id}:
{e}[/red]", file=sys.stderr)
+ rich.print(
+ f"[red]Error while trying to {operation} Dag {escape(dag_id)}:
{escape(str(e))}[/red]",
+ file=sys.stderr,
+ )
sys.exit(1)
response_dict = response.model_dump(mode="json")
@@ -116,7 +120,7 @@ def next_execution(args, api_client=NEW_API_CLIENT) -> dict
| None:
try:
response = api_client.dags.get(dag_id=args.dag_id)
except ServerResponseError as e:
- rich.print(f"[red]Error retrieving Dag {args.dag_id}: {e}[/red]")
+ rich.print(f"[red]Error retrieving Dag {escape(args.dag_id)}:
{escape(str(e))}[/red]")
sys.exit(1)
response_dict = response.model_dump(mode="json")
@@ -244,7 +248,7 @@ def _print_dag_runs_to_clear(dag_id: str, dag_runs: list)
-> None:
partition_date = getattr(dag_run, "partition_date", None)
partition_date_display = partition_date.isoformat() if partition_date
is not None else "-"
rich.print(
- f" - {dag_run.dag_run_id} (logical date: {logical_date},
partition date: {partition_date_display})"
+ f" - {escape(dag_run.dag_run_id)} (logical date: {logical_date},
partition date: {partition_date_display})"
)
diff --git a/airflow-ctl/src/airflowctl/ctl/commands/task_command.py
b/airflow-ctl/src/airflowctl/ctl/commands/task_command.py
index dc756bd8056..31bf6dfcb84 100644
--- a/airflow-ctl/src/airflowctl/ctl/commands/task_command.py
+++ b/airflow-ctl/src/airflowctl/ctl/commands/task_command.py
@@ -21,6 +21,7 @@ import sys
from typing import TYPE_CHECKING
import rich
+from rich.markup import escape
from airflowctl.api.client import NEW_API_CLIENT, ClientKind,
ServerResponseError, provide_api_client
from airflowctl.api.datamodels.generated import StructuredLogMessage,
TaskInstanceState
@@ -99,7 +100,7 @@ def failed_deps(args, api_client=NEW_API_CLIENT) -> None:
except ServerResponseError as e:
if e.response.status_code == 404:
rich.print(
- f"[red]{_task_instance_not_found_message(args.dag_id, run_id,
args.task_id, args.map_index, e)}[/red]"
+ f"[red]{escape(_task_instance_not_found_message(args.dag_id,
run_id, args.task_id, args.map_index, e))}[/red]"
)
sys.exit(1)
raise
@@ -122,7 +123,9 @@ def states_for_dag_run(args, api_client=NEW_API_CLIENT) ->
None:
task_instances = api_client.task_instances.list(dag_id=args.dag_id,
dag_run_id=run_id).task_instances
except ServerResponseError as e:
if e.response.status_code == 404:
- rich.print(f"[red]Dag run {run_id!r} of Dag {args.dag_id!r} not
found[/red]")
+ rich.print(
+ f"[red]Dag run {escape(repr(run_id))} of Dag
{escape(repr(args.dag_id))} not found[/red]"
+ )
sys.exit(1)
raise
@@ -150,7 +153,7 @@ def state(args, api_client=NEW_API_CLIENT) -> None:
except ServerResponseError as e:
if e.response.status_code == 404:
rich.print(
- f"[red]{_task_instance_not_found_message(args.dag_id, run_id,
args.task_id, args.map_index, e)}[/red]"
+ f"[red]{escape(_task_instance_not_found_message(args.dag_id,
run_id, args.task_id, args.map_index, e))}[/red]"
)
sys.exit(1)
raise
@@ -175,7 +178,7 @@ def logs(args, api_client=NEW_API_CLIENT) -> None:
except ServerResponseError as e:
if e.response.status_code == 404:
rich.print(
- f"[red]{_task_instance_not_found_message(args.dag_id,
args.dag_run_id, args.task_id, args.map_index, e)}[/red]"
+ f"[red]{escape(_task_instance_not_found_message(args.dag_id,
args.dag_run_id, args.task_id, args.map_index, e))}[/red]"
)
sys.exit(1)
raise
diff --git a/airflow-ctl/src/airflowctl/ctl/utils/dag_run.py
b/airflow-ctl/src/airflowctl/ctl/utils/dag_run.py
index b2bdad4af38..b44a26b463a 100644
--- a/airflow-ctl/src/airflowctl/ctl/utils/dag_run.py
+++ b/airflow-ctl/src/airflowctl/ctl/utils/dag_run.py
@@ -23,6 +23,7 @@ import sys
from typing import TYPE_CHECKING
import rich
+from rich.markup import escape
from airflowctl.api.client import ServerResponseError
@@ -45,7 +46,7 @@ def _get_dag_run_by_run_id(api_client: Client, dag_id: str,
run_id: str) -> DAGR
except ServerResponseError as e:
if e.response.status_code != 404:
raise
- rich.print(f"[red]Dag run {run_id!r} of Dag {dag_id!r} not
found[/red]")
+ rich.print(f"[red]Dag run {escape(repr(run_id))} of Dag
{escape(repr(dag_id))} not found[/red]")
sys.exit(1)
@@ -54,7 +55,7 @@ def _get_dag_run_by_logical_date(api_client: Client, dag_id:
str, value: str) ->
try:
logical_date = datetime.datetime.fromisoformat(value.replace("Z",
"+00:00"))
except ValueError:
- rich.print(f"[red]Invalid --logical-date: {value!r}[/red]")
+ rich.print(f"[red]Invalid --logical-date: {escape(repr(value))}[/red]")
sys.exit(1)
if logical_date.tzinfo is None:
rich.print("[red]--logical-date must include a timezone offset[/red]")
@@ -74,7 +75,9 @@ def _get_dag_run_by_logical_date(api_client: Client, dag_id:
str, value: str) ->
if e.response.status_code != 404:
raise
if not dag_runs:
- rich.print(f"[red]Dag run for {dag_id} with logical date {value!r} not
found[/red]")
+ rich.print(
+ f"[red]Dag run for {escape(dag_id)} with logical date
{escape(repr(value))} not found[/red]"
+ )
sys.exit(1)
return dag_runs[0]
diff --git a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_auth_command.py
b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_auth_command.py
index 2856394ac40..11a30dd3741 100644
--- a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_auth_command.py
+++ b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_auth_command.py
@@ -24,7 +24,7 @@ from unittest.mock import patch
import pytest
-from airflowctl.api.client import ClientKind
+from airflowctl.api.client import Client, ClientKind
from airflowctl.api.datamodels.auth_generated import LoginResponse
from airflowctl.ctl import cli_parser
from airflowctl.ctl.commands import auth_command
@@ -647,3 +647,53 @@ class TestListEnvs:
auth_command.list_envs(args)
mock_get_password.assert_called_once_with("airflowctl",
"api_token_prod.json.region")
+
+
+def test_login_prints_failure_containing_markup(capsys):
+ api_client = mock.MagicMock(spec=Client)
+ api_client.login.login_with_username_and_password.side_effect =
RuntimeError("bad [/x]")
+
+ with pytest.raises(SystemExit) as exc_info:
+ auth_command.login(
+ cli_parser.get_parser().parse_args(
+ [
+ "auth",
+ "login",
+ "--api-url",
+ "http://localhost:8080";,
+ "--username",
+ "user",
+ "--password",
+ "wrong",
+ ]
+ ),
+ api_client=api_client,
+ )
+
+ assert exc_info.value.code == 1
+ assert "bad [/x]" in capsys.readouterr().out
+
+
+def test_token_prints_failure_containing_markup(capsys):
+ api_client = mock.MagicMock(spec=Client)
+ api_client.login.login_with_username_and_password.side_effect =
RuntimeError("bad [/x]")
+
+ with pytest.raises(SystemExit) as exc_info:
+ auth_command.get_token(
+ cli_parser.get_parser().parse_args(
+ [
+ "auth",
+ "token",
+ "--api-url",
+ "http://localhost:8080";,
+ "--username",
+ "user",
+ "--password",
+ "wrong",
+ ]
+ ),
+ api_client=api_client,
+ )
+
+ assert exc_info.value.code == 1
+ assert "bad [/x]" in capsys.readouterr().err
diff --git a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_config_command.py
b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_config_command.py
index d7b7662e4ce..d7d68b8aee6 100644
--- a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_config_command.py
+++ b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_config_command.py
@@ -17,11 +17,12 @@
from __future__ import annotations
import os
+from unittest import mock
from unittest.mock import patch
import pytest
-from airflowctl.api.client import ClientKind
+from airflowctl.api.client import Client, ClientKind
from airflowctl.api.datamodels.generated import Config, ConfigOption,
ConfigSection
from airflowctl.ctl import cli_parser
from airflowctl.ctl.commands import config_command
@@ -611,3 +612,14 @@ class TestCliConfigCommands:
assert "fernet_key" in output_str
assert "< hidden >" in output_str
assert "sql_alchemy_conn" in output_str
+
+
+def test_lint_prints_failure_containing_markup(capsys):
+ api_client = mock.MagicMock(spec=Client)
+ api_client.configs.list.side_effect = RuntimeError("bad [/x]")
+
+ with pytest.raises(SystemExit) as exc_info:
+ config_command.lint(cli_parser.get_parser().parse_args(["config",
"lint"]), api_client=api_client)
+
+ assert exc_info.value.code == 1
+ assert "bad [/x]" in capsys.readouterr().out
diff --git
a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_connections_command.py
b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_connections_command.py
index bd0658cab9f..ecbe9daabb8 100644
--- a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_connections_command.py
+++ b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_connections_command.py
@@ -323,3 +323,28 @@ class TestCliConnectionCommands:
bulk_body = mock_client.connections.bulk.call_args[0][0]
action = bulk_body.actions[0]
assert action.action_on_existence == expected_enum
+
+
[email protected](
+ "bulk_setup",
+ [
+ pytest.param(
+ lambda bulk: setattr(bulk.return_value.create, "errors", ["bad
[/x]"]), id="bulk-errors"
+ ),
+ pytest.param(lambda bulk: setattr(bulk, "side_effect",
RuntimeError("bad [/x]")), id="exception"),
+ ],
+)
+def test_import_prints_failure_containing_markup(bulk_setup, tmp_path, capsys):
+ connections_file = tmp_path / "connections.json"
+ connections_file.write_text(json.dumps({"my_conn": {"conn_type": "http"}}))
+ api_client = mock.MagicMock(spec=Client)
+ bulk_setup(api_client.connections.bulk)
+
+ with pytest.raises(SystemExit) as exc_info:
+ connection_command.import_(
+ cli_parser.get_parser().parse_args(["connections", "import",
connections_file.as_posix()]),
+ api_client=api_client,
+ )
+
+ assert exc_info.value.code == 1
+ assert "bad [/x]" in capsys.readouterr().out
diff --git a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_dag_command.py
b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_dag_command.py
index b9957333314..b8b64021b4b 100644
--- a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_dag_command.py
+++ b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_dag_command.py
@@ -25,7 +25,7 @@ from unittest.mock import Mock, call, create_autospec
import httpx
import pytest
-from airflowctl.api.client import ClientKind
+from airflowctl.api.client import Client, ClientKind
from airflowctl.api.datamodels.generated import ClearTaskInstancesBody,
DAGResponse, DagSchedulingState
from airflowctl.api.operations import DagRunOperations, ServerResponseError,
TasksOperations
from airflowctl.ctl import cli_parser
@@ -228,6 +228,33 @@ class TestDagCommands:
api_client=api_client,
)
+ def test_pause_fail_prints_server_error_containing_markup(self,
api_client_maker, capsys):
+ api_client = api_client_maker(
+ path=f"/api/v2/dags/{self.dag_id}",
+ response_json={"detail": "bad [/x] value"},
+ expected_http_status_code=404,
+ kind=ClientKind.CLI,
+ )
+ with pytest.raises(SystemExit):
+ dag_command.pause(
+ self.parser.parse_args(["dags", "pause", self.dag_id]),
+ api_client=api_client,
+ )
+
+ assert "[/x]" in capsys.readouterr().err
+
+ def test_pause_fail_prints_dag_id_containing_markup(self, capsys):
+ api_client = Mock(spec=Client)
+ api_client.dags.update.side_effect = _make_server_error(404)
+
+ with pytest.raises(SystemExit):
+ dag_command.pause(
+ self.parser.parse_args(["dags", "pause", "my_dag[/y]"]),
+ api_client=api_client,
+ )
+
+ assert "my_dag[/y]" in capsys.readouterr().err
+
def test_unpause_dag(self, api_client_maker, monkeypatch):
api_client = api_client_maker(
path=f"/api/v2/dags/{self.dag_id}",
@@ -364,6 +391,33 @@ class TestDagCommands:
)
assert result is None
+ def test_next_execution_fail_prints_dag_id_containing_markup(self, capsys):
+ api_client = Mock(spec=Client)
+ api_client.dags.get.side_effect = _make_server_error(404)
+
+ with pytest.raises(SystemExit):
+ dag_command.next_execution(
+ self.parser.parse_args(["dags", "next-execution",
"my_dag[/y]"]),
+ api_client=api_client,
+ )
+
+ assert "my_dag[/y]" in capsys.readouterr().out
+
+ def test_next_execution_fail_prints_server_error_containing_markup(self,
api_client_maker, capsys):
+ api_client = api_client_maker(
+ path=f"/api/v2/dags/{self.dag_id}",
+ response_json={"detail": "bad [/x] value"},
+ expected_http_status_code=404,
+ kind=ClientKind.CLI,
+ )
+ with pytest.raises(SystemExit):
+ dag_command.next_execution(
+ self.parser.parse_args(["dags", "next-execution",
self.dag_id]),
+ api_client=api_client,
+ )
+
+ assert "[/x]" in capsys.readouterr().out
+
def test_next_execution_fail(self, api_client_maker):
api_client = api_client_maker(
path=f"/api/v2/dags/{self.dag_id}",
@@ -730,6 +784,19 @@ class TestDagCommands:
assert result == {"dag_run_count": 1, "cleared_task_instances": 0,
"cancelled": True}
api_client.tasks.clear.assert_not_called()
+ @pytest.mark.parametrize("run_id", ["run_[/x]", "run[abc]"])
+ def test_clear_prompt_lists_run_id_containing_markup(self, run_id,
monkeypatch, capsys):
+ api_client = self._api_client_mock()
+ api_client.dag_runs.get.return_value = self._dag_run(run_id)
+ monkeypatch.setattr("builtins.input", lambda _: "n")
+
+ dag_command.clear(
+ self.parser.parse_args(["dags", "clear", self.dag_id, "--run-id",
run_id]),
+ api_client=api_client,
+ )
+
+ assert f"- {run_id} (" in capsys.readouterr().out
+
@pytest.mark.parametrize(
"command",
[
diff --git a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_task_command.py
b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_task_command.py
index c803a880221..067a453ebe6 100644
--- a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_task_command.py
+++ b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_task_command.py
@@ -23,6 +23,7 @@ from unittest import mock
import httpx
import pytest
+from airflowctl.api.client import Client
from airflowctl.api.datamodels.generated import (
StructuredLogMessage,
TaskDependencyCollectionResponse,
@@ -505,6 +506,20 @@ class TestStatesForDagRun:
output="json",
)
+ def test_states_for_dag_run_not_found_prints_ids_containing_markup(self,
capsys):
+ api_client = mock.MagicMock(spec=Client)
+ api_client.task_instances.list.side_effect = _make_server_error(404)
+
+ with pytest.raises(SystemExit, match="1"):
+ task_command.states_for_dag_run(
+ self.parser.parse_args(["tasks", "states-for-dag-run",
"my_dag[/y]", "run_[/x]"]),
+ api_client=api_client,
+ )
+
+ output = capsys.readouterr().out
+ assert "run_[/x]" in output
+ assert "my_dag[/y]" in output
+
@mock.patch("airflowctl.ctl.commands.task_command.AirflowConsole")
def test_states_for_dag_run_by_logical_date(self, mock_console_cls):
api_client =
self._make_api_client([self._make_task_instance("task_a")])
@@ -1030,3 +1045,36 @@ class TestLogs:
),
api_client=api_client,
)
+
+
[email protected](
+ ("command", "argv", "api_method"),
+ [
+ pytest.param(
+ task_command.failed_deps,
+ ["tasks", "failed-deps", "my_dag", "my_task[/x]", "my_run"],
+ "get_dependencies",
+ id="failed-deps",
+ ),
+ pytest.param(
+ task_command.state,
+ ["tasks", "state", "my_dag", "my_task[/x]", "my_run"],
+ "get",
+ id="state",
+ ),
+ pytest.param(
+ task_command.logs,
+ ["tasks", "logs", "my_dag", "my_run", "my_task[/x]",
"--try-number", "1"],
+ "logs",
+ id="logs",
+ ),
+ ],
+)
+def test_task_instance_not_found_prints_task_id_containing_markup(command,
argv, api_method, capsys):
+ api_client = mock.MagicMock(spec=Client)
+ getattr(api_client.task_instances, api_method).side_effect =
_make_server_error(404)
+
+ with pytest.raises(SystemExit, match="1"):
+ command(cli_parser.get_parser().parse_args(argv),
api_client=api_client)
+
+ assert "'my_task[/x]'" in _normalize_rich_output(capsys.readouterr().out)
diff --git a/airflow-ctl/tests/airflow_ctl/ctl/utils/test_dag_run.py
b/airflow-ctl/tests/airflow_ctl/ctl/utils/test_dag_run.py
index eeefcf2036e..b053747056c 100644
--- a/airflow-ctl/tests/airflow_ctl/ctl/utils/test_dag_run.py
+++ b/airflow-ctl/tests/airflow_ctl/ctl/utils/test_dag_run.py
@@ -94,6 +94,25 @@ def test_get_by_run_id_404_exits(capsys):
assert "not found" in capsys.readouterr().out
+def test_get_by_run_id_404_prints_run_id_containing_markup(capsys):
+ client = Mock(spec=Client)
+ client.dag_runs.get.side_effect = server_error(404)
+
+ with pytest.raises(SystemExit):
+ _get_dag_run_by_run_id(client, "my_dag[/y]", "run_[/x]")
+
+ output = capsys.readouterr().out
+ assert "run_[/x]" in output
+ assert "my_dag[/y]" in output
+
+
+def test_get_by_logical_date_prints_invalid_value_containing_markup(capsys):
+ with pytest.raises(SystemExit):
+ _get_dag_run_by_logical_date(Mock(spec=Client), "my_dag", "[/x]")
+
+ assert "[/x]" in capsys.readouterr().out
+
+
def test_get_by_run_id_other_error_propagates():
client = Mock(spec=Client)
client.dag_runs.get.side_effect = server_error(500)
@@ -145,6 +164,16 @@ def test_get_by_logical_date_not_found_exits(capsys):
assert "not found" in capsys.readouterr().out
+def test_get_by_logical_date_not_found_prints_dag_id_containing_markup(capsys):
+ client = Mock(spec=Client)
+ client.dag_runs.list.return_value = Mock(dag_runs=[])
+
+ with pytest.raises(SystemExit):
+ _get_dag_run_by_logical_date(client, "my_dag[/y]",
"2026-01-01T00:00:00Z")
+
+ assert "my_dag[/y]" in capsys.readouterr().out
+
+
def test_resolve_dag_run_id_skips_api_for_run_id():
client = Mock(spec=Client)
