diff --git a/src/onepin/_cli/_dispatch.py b/src/onepin/_cli/_dispatch.py index f7332d0..637e522 100644 --- a/src/onepin/_cli/_dispatch.py +++ b/src/onepin/_cli/_dispatch.py @@ -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: @@ -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") diff --git a/src/onepin/_cli/_spec.py b/src/onepin/_cli/_spec.py index 9b88476..48a305f 100644 --- a/src/onepin/_cli/_spec.py +++ b/src/onepin/_cli/_spec.py @@ -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. @@ -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, ...]: @@ -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", diff --git a/tests/build/test_sdk_contract.py b/tests/build/test_sdk_contract.py index 84241e9..ff56d14 100644 --- a/tests/build/test_sdk_contract.py +++ b/tests/build/test_sdk_contract.py @@ -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 @@ -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()) @@ -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} diff --git a/tests/unit/test_dispatch.py b/tests/unit/test_dispatch.py index 6ddf0c3..5ce7792 100644 --- a/tests/unit/test_dispatch.py +++ b/tests/unit/test_dispatch.py @@ -6,6 +6,7 @@ import enum import json from pathlib import Path +from types import SimpleNamespace import pytest @@ -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")