Skip to content
Open
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
29 changes: 24 additions & 5 deletions src/yertle/sre/cli/status.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@
import os
from dataclasses import dataclass

import yertle
from yertle.shared import auth
from yertle.sre.tools._shell import run_cli

PROBE_TIMEOUT = 5
Expand All @@ -37,11 +39,28 @@ def probe_anthropic() -> ProbeResult:


def probe_yertle() -> ProbeResult:
"""Check the yertle CLI by listing orgs."""
result = run_cli(["yertle", "orgs", "--format", "json"], timeout=PROBE_TIMEOUT)
if not result.ok:
return ProbeResult("yertle", False, "not authenticated — run `yertle login`")
return ProbeResult("yertle", True, "authenticated")
"""Check Yertle credentials by listing organizations.

Calls the SDK in-process rather than shelling out to the `yertle` CLI.
The CLI form was `yertle orgs --format json`, which silently became a
usage error the moment `orgs` grew subcommands — and because the test
mocked the subprocess, nothing caught it; the probe just started telling
authenticated users to log in.

Going through the SDK deletes the string that can drift and answers the
question the probe actually asks — are these credentials good — without a
process spawn. `aws` and `gh` still shell out because they genuinely are
external CLIs.
"""
try:
api_url = auth.resolve().api_url
yertle.orgs.list()
except auth.AuthError as e:
# AuthError's message is written to be shown to a user as-is.
return ProbeResult("yertle", False, str(e))
except Exception as e: # noqa: BLE001 — probes never raise; they report.
return ProbeResult("yertle", False, f"could not reach the API ({type(e).__name__})")
return ProbeResult("yertle", True, f"authenticated to {api_url}")


def probe_aws() -> ProbeResult:
Expand Down
46 changes: 38 additions & 8 deletions tests/sre/test_status.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,13 @@

from __future__ import annotations

import datetime
from unittest.mock import patch

from yertle_client.models import OrganizationListResponse, OrganizationResponse

from tests.sre.conftest import FakeCompleted
from yertle.shared import auth as auth_mod
from yertle.sre.cli.status import (
probe_all,
probe_anthropic,
Expand All @@ -12,6 +18,18 @@
)


def _orgs_response() -> OrganizationListResponse:
now = datetime.datetime(2026, 9, 2, tzinfo=datetime.UTC)
org = OrganizationResponse(
id="org-1",
name="Acme",
public_id="acme",
created_at=now,
updated_at=now,
)
return OrganizationListResponse(organizations=[org], total=1)


def test_probe_anthropic_set(monkeypatch):
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
result = probe_anthropic()
Expand All @@ -26,20 +44,32 @@ def test_probe_anthropic_unset(monkeypatch):
assert "ANTHROPIC_API_KEY" in result.detail


def test_probe_yertle_success(fake_cli):
fake_cli(lambda _argv: FakeCompleted(stdout="[]", stderr="", returncode=0))
@patch("yertle.orgs.list_organizations_orgs_get.sync", return_value=_orgs_response())
@patch("yertle._client.get_client", return_value=object())
def test_probe_yertle_success(_get_client, _sync, monkeypatch):
monkeypatch.setenv("YERTLE_API_URL", "https://api.example.test")
monkeypatch.setenv("YERTLE_TOKEN", "yrt_test")
result = probe_yertle()
assert result.ok
assert result.detail == "authenticated"
assert "api.example.test" in result.detail


def test_probe_yertle_failure(fake_cli):
fake_cli(
lambda _argv: FakeCompleted(stdout="", stderr="auth: not logged in", returncode=1),
)
def test_probe_yertle_reports_missing_credentials(monkeypatch, tmp_path):
"""No token must read as unauthenticated, not as a crash."""
monkeypatch.delenv("YERTLE_TOKEN", raising=False)
monkeypatch.setattr(auth_mod, "CONFIG_PATH", tmp_path / "config.json")
result = probe_yertle()
assert not result.ok


@patch("yertle.orgs.list_organizations_orgs_get.sync", side_effect=ConnectionError("boom"))
@patch("yertle._client.get_client", return_value=object())
def test_probe_yertle_reports_an_unreachable_api(_get_client, _sync, monkeypatch):
"""Probes never raise — a dead backend is a result, not an exception."""
monkeypatch.setenv("YERTLE_TOKEN", "yrt_test")
result = probe_yertle()
assert not result.ok
assert "yertle login" in result.detail
assert "ConnectionError" in result.detail


def test_probe_aws_success_extracts_arn(fake_cli, monkeypatch):
Expand Down
27 changes: 26 additions & 1 deletion tests/sre/test_tools_yertle.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,8 @@
from __future__ import annotations

from tests.sre.conftest import FakeCompleted
from yertle.sre.tools.yertle import yertle_run
from yertle.cli.main import app as cli_app
from yertle.sre.tools.yertle import YERTLE_READ_COMMANDS, yertle_run


def test_yertle_run_allows_listed_commands(fake_cli):
Expand Down Expand Up @@ -81,3 +82,27 @@ def test_yertle_run_translates_failure(fake_cli):
out = yertle_run.invoke({"argv": ["orgs", "list"]})
assert out.startswith("yertle CLI failed:")
assert "not found" in out


def test_allowlist_only_names_commands_the_cli_actually_has():
"""The allowlist is a hand-maintained mirror of the CLI, so it can rot.

It already did once: it was copied from the Go CLI and listed `nodes`,
`tree`, `canvas`, `about` and `config`, none of which the Python CLI had —
so the agent was told to call five commands that could only fail. The
subprocess mock in these tests hid it, because a fake `run_cli` happily
"succeeds" for a command that does not exist.

This asserts against the Typer app itself, which cannot drift.
"""
registered = {command.name for command in cli_app.registered_commands}
registered |= {
group.name or (group.typer_instance.info.name if group.typer_instance else None)
for group in cli_app.registered_groups
}

unknown = YERTLE_READ_COMMANDS - registered
assert not unknown, (
f"allowlisted commands the CLI does not have: {sorted(unknown)}. "
f"Registered: {sorted(n for n in registered if n)}"
)
Loading