Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 51 additions & 1 deletion cueapi/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -743,10 +743,48 @@ def executions() -> None:
@executions.command(name="list")
@click.option("--cue-id", "cue_id", default=None, help="Filter to a specific cue")
@click.option("--status", default=None, help="Filter by execution status")
@click.option(
"--outcome-state",
"outcome_state",
default=None,
help=(
"Filter by outcome_state: reported_success / reported_failure / "
"verified_success / verification_pending / verification_failed / unknown."
),
)
@click.option(
"--result-type",
"result_type",
default=None,
help="Filter by evidence result_type (e.g. 'pr', 'issue', 'comment').",
)
@click.option(
"--has-evidence",
"has_evidence",
is_flag=True,
default=False,
help="Filter to executions that reported evidence (evidence_external_id is set).",
)
@click.option(
"--triggered-by",
"triggered_by",
default=None,
help="Filter by triggered_by: scheduled / manual_fire / chain.",
)
@click.option("--limit", default=20, type=int, help="Max results")
@click.option("--offset", default=0, type=int, help="Offset for pagination")
@click.pass_context
def executions_list(ctx: click.Context, cue_id: Optional[str], status: Optional[str], limit: int, offset: int) -> None:
def executions_list(
ctx: click.Context,
cue_id: Optional[str],
status: Optional[str],
outcome_state: Optional[str],
result_type: Optional[str],
has_evidence: bool,
triggered_by: Optional[str],
limit: int,
offset: int,
) -> None:
"""List historical executions across all cues."""
try:
with CueAPIClient(api_key=ctx.obj.get("api_key"), profile=ctx.obj.get("profile")) as client:
Expand All @@ -755,6 +793,18 @@ def executions_list(ctx: click.Context, cue_id: Optional[str], status: Optional[
params["cue_id"] = cue_id
if status:
params["status"] = status
if outcome_state:
params["outcome_state"] = outcome_state
if result_type:
params["result_type"] = result_type
# Server-side `has_evidence` filter is meaningful only when True
# (it ANDs `evidence_external_id IS NOT NULL`). Unset = no filter,
# so omit from query params when False rather than send `false`
# which would still mean the same thing but adds URL noise.
if has_evidence:
params["has_evidence"] = "true"
if triggered_by:
params["triggered_by"] = triggered_by
resp = client.get("/executions", params=params)
if resp.status_code != 200:
echo_error(f"Failed (HTTP {resp.status_code})")
Expand Down
124 changes: 124 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -1409,3 +1409,127 @@ def get(self, path, **_):
result = runner.invoke(main, ["executions", "get", "exec_abc"])
assert result.exit_code == 0, result.output
assert "Payload:" not in result.output


# --- executions list filter parity (cueapi-cli #25 manifest gap) ---


class _FakeResp:
def __init__(self, status_code: int, payload: dict):
self.status_code = status_code
self._payload = payload

def json(self):
return self._payload


class _ListClient:
def __init__(self):
self.last_params: Optional[dict] = None

def __enter__(self):
return self

def __exit__(self, *_):
pass

def get(self, path, params=None, **_):
self.last_params = params
return _FakeResp(200, {"executions": [], "total": 0, "limit": 20, "offset": 0})


def _patched_list_client(monkeypatch, holder):
import cueapi.cli as cli_mod

def fake_factory(*_, **__):
holder["client"] = _ListClient()
return holder["client"]

monkeypatch.setattr(cli_mod, "CueAPIClient", fake_factory)


def test_executions_list_help_includes_new_filters():
result = runner.invoke(main, ["executions", "list", "--help"])
assert result.exit_code == 0
assert "--outcome-state" in result.output
assert "--result-type" in result.output
assert "--has-evidence" in result.output
assert "--triggered-by" in result.output


def test_executions_list_outcome_state_passed_as_query_param(monkeypatch):
holder: dict = {}
_patched_list_client(monkeypatch, holder)
result = runner.invoke(
main,
["executions", "list", "--outcome-state", "verified_success"],
)
assert result.exit_code == 0, result.output
assert holder["client"].last_params.get("outcome_state") == "verified_success"


def test_executions_list_result_type_passed(monkeypatch):
holder: dict = {}
_patched_list_client(monkeypatch, holder)
result = runner.invoke(
main,
["executions", "list", "--result-type", "pr"],
)
assert result.exit_code == 0, result.output
assert holder["client"].last_params.get("result_type") == "pr"


def test_executions_list_has_evidence_only_sent_when_true(monkeypatch):
# has_evidence is a flag — present means True. Unset = omit. Pinning
# this so a refactor can't silently start sending `false` (which would
# still mean "no filter" server-side, but creates noisy URLs and invites
# future bugs).
holder: dict = {}
_patched_list_client(monkeypatch, holder)
result_no_flag = runner.invoke(main, ["executions", "list"])
assert result_no_flag.exit_code == 0
assert "has_evidence" not in (holder["client"].last_params or {})

holder2: dict = {}
_patched_list_client(monkeypatch, holder2)
result_with_flag = runner.invoke(main, ["executions", "list", "--has-evidence"])
assert result_with_flag.exit_code == 0
assert holder2["client"].last_params.get("has_evidence") == "true"


def test_executions_list_triggered_by_passed(monkeypatch):
holder: dict = {}
_patched_list_client(monkeypatch, holder)
result = runner.invoke(
main,
["executions", "list", "--triggered-by", "manual_fire"],
)
assert result.exit_code == 0, result.output
assert holder["client"].last_params.get("triggered_by") == "manual_fire"


def test_executions_list_combines_all_filters(monkeypatch):
holder: dict = {}
_patched_list_client(monkeypatch, holder)
result = runner.invoke(
main,
[
"executions", "list",
"--cue-id", "cue_xyz",
"--status", "success",
"--outcome-state", "verified_success",
"--result-type", "pr",
"--has-evidence",
"--triggered-by", "scheduled",
"--limit", "50",
],
)
assert result.exit_code == 0, result.output
p = holder["client"].last_params
assert p["cue_id"] == "cue_xyz"
assert p["status"] == "success"
assert p["outcome_state"] == "verified_success"
assert p["result_type"] == "pr"
assert p["has_evidence"] == "true"
assert p["triggered_by"] == "scheduled"
assert p["limit"] == 50
Loading