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
28 changes: 21 additions & 7 deletions src/onepin/_cli/_dispatch.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,12 +55,26 @@ def _annotation_for(opt: Opt) -> Any:
return Optional[str]


def _resolve_method(client: Any, dotted: str) -> Callable[..., Any]:
"""Resolve a dotted method path (e.g. ``workflows.runs.start``) against a built client."""
obj: Any = client
for part in dotted.split("."):
obj = getattr(obj, part)
return obj # type: ignore[no-any-return]
def _resolve_method(client: Any, paths: str | tuple[str, ...]) -> Callable[..., Any]:
"""Resolve the first available dotted SDK method path against a built client."""
candidates = (paths,) if isinstance(paths, str) else paths
last_error: AttributeError | None = None
for dotted in candidates:
obj: Any = client
try:
for part in dotted.split("."):
obj = getattr(obj, part)
except AttributeError as exc:
try:
inspect.getattr_static(obj, part)
except AttributeError:
last_error = exc
continue
raise
return obj # type: ignore[no-any-return]

attempted = ", ".join(repr(path) for path in candidates)
raise AttributeError(f"Could not resolve SDK method; attempted paths: {attempted}") from last_error


def _coerce_value(value: Any) -> Any:
Expand Down Expand Up @@ -411,7 +425,7 @@ def _run(cmd: Cmd, bound: dict[str, Any]) -> None:
_confirm_destructive(cmd, bool(bound.get("yes", False)), json_on=json_on)

client = get_client()
method = _resolve_method(client, cmd.method)
method = _resolve_method(client, cmd.method_paths)
positional, kwargs = _build_kwargs(cmd, bound)
if _accepts_workspace_kwarg(method):
workspace = _state.root_options.get("workspace")
Expand Down
10 changes: 9 additions & 1 deletion src/onepin/_cli/_spec.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ class Cmd:
name: Command name within the group/subgroup.
method: Dotted SDK method path relative to the client, e.g. ``workflows.list`` or
``workflows.runs.start``.
fallback_methods: Transitional SDK method paths tried in order when ``method`` is absent.
help: Command help text.
subgroup: Optional nested sub-Typer (``runs``, ``members``, ``stats``).
args: Positional arguments, ``[(dest, help)]``; forwarded positionally in order.
Expand All @@ -85,6 +86,12 @@ class Cmd:
success_msg: Optional[str] = None
destructive: bool = False
redact: bool = False
fallback_methods: tuple[str, ...] = ()

@property
def method_paths(self) -> tuple[str, ...]:
"""SDK method paths in resolution order, canonical first."""
return (self.method, *self.fallback_methods)

@property
def path(self) -> tuple[str, ...]:
Expand Down Expand Up @@ -283,13 +290,14 @@ def _list_opts(*extra: Opt) -> list[Opt]:
Cmd(
"workflows",
"steps",
"workflows.get_run_steps",
"workflows.runs.steps",
"List the steps of a run.",
subgroup="runs",
args=[("workflow_id", "Workflow UUID."), ("run_id", "Run UUID.")],
options=[_JSON],
unwrap="list",
columns=[],
fallback_methods=("workflows.get_run_steps",),
),
Cmd(
"workflows",
Expand Down
14 changes: 6 additions & 8 deletions tests/build/test_sdk_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

import pytest

from onepin._cli._dispatch import _resolve_method
from onepin._cli._spec import TABLE, Cmd
from onepin.client import OnePinClient

Expand All @@ -25,23 +26,20 @@
_LOCAL_FLAGS = {"--json", "--reveal", "--yes"}


def _resolve_method(dotted: str):
def _resolve_cmd_method(cmd: Cmd):
client = OnePinClient(token="op_live_test")
obj = client
for part in dotted.split("."):
obj = getattr(obj, part)
return obj
return _resolve_method(client, cmd.method_paths)


@pytest.mark.parametrize("cmd", TABLE, ids=lambda c: ".".join(c.path))
def test_method_resolves(cmd: Cmd) -> None:
method = _resolve_method(cmd.method)
method = _resolve_cmd_method(cmd)
assert callable(method), f"{cmd.method} is not callable"


@pytest.mark.parametrize("cmd", TABLE, ids=lambda c: ".".join(c.path))
def test_declared_params_subset_of_signature(cmd: Cmd) -> None:
method = _resolve_method(cmd.method)
method = _resolve_cmd_method(cmd)
sig = inspect.signature(method)
params = sig.parameters
accepts_kwargs = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values())
Expand All @@ -62,7 +60,7 @@ def test_declared_params_subset_of_signature(cmd: Cmd) -> None:

@pytest.mark.parametrize("cmd", TABLE, ids=lambda c: ".".join(c.path))
def test_required_params_supplied(cmd: Cmd) -> None:
method = _resolve_method(cmd.method)
method = _resolve_cmd_method(cmd)
sig = inspect.signature(method)

supplied = {dest for dest, _ in cmd.args}
Expand Down
70 changes: 70 additions & 0 deletions tests/unit/test_dispatch.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import enum
import json
from pathlib import Path
from types import SimpleNamespace

import pytest

Expand All @@ -14,6 +15,75 @@
from onepin._cli._spec import Cmd, Opt


class TestMethodResolution:
def test_resolves_canonical_path(self) -> None:
canonical = object()
client = SimpleNamespace(workflows=SimpleNamespace(runs=SimpleNamespace(steps=canonical)))

assert _dispatch._resolve_method(client, ("workflows.runs.steps",)) is canonical

def test_falls_back_when_canonical_path_is_absent(self) -> None:
legacy = object()
client = SimpleNamespace(workflows=SimpleNamespace(get_run_steps=legacy))

resolved = _dispatch._resolve_method(
client,
("workflows.runs.steps", "workflows.get_run_steps"),
)

assert resolved is legacy

def test_prefers_canonical_path_when_both_exist(self) -> None:
canonical = object()
legacy = object()
client = SimpleNamespace(
workflows=SimpleNamespace(
runs=SimpleNamespace(steps=canonical),
get_run_steps=legacy,
)
)

resolved = _dispatch._resolve_method(
client,
("workflows.runs.steps", "workflows.get_run_steps"),
)

assert resolved is canonical

def test_does_not_fall_back_when_canonical_attribute_raises(self) -> None:
class BrokenRuns:
@property
def steps(self) -> object:
raise AttributeError("canonical initialization failed")

legacy = object()
client = SimpleNamespace(
workflows=SimpleNamespace(
runs=BrokenRuns(),
get_run_steps=legacy,
)
)

with pytest.raises(AttributeError, match="canonical initialization failed"):
_dispatch._resolve_method(
client,
("workflows.runs.steps", "workflows.get_run_steps"),
)

def test_reports_all_attempted_paths_when_none_resolve(self) -> None:
client = SimpleNamespace(workflows=SimpleNamespace())

with pytest.raises(AttributeError) as exc_info:
_dispatch._resolve_method(
client,
("workflows.runs.steps", "workflows.get_run_steps"),
)

message = str(exc_info.value)
assert "workflows.runs.steps" in message
assert "workflows.get_run_steps" in message


class TestTransforms:
def test_wrap_list(self) -> None:
opt = Opt("--sort", ("name",), None, transform="wrap_list")
Expand Down