diff --git a/README.md b/README.md index f9525b6d..5fcf2453 100644 --- a/README.md +++ b/README.md @@ -224,6 +224,8 @@ graphiant invoke v1_edges_summary_get --kwargs '{"enterprise_id": 123}' ```bash graphiant invoke v1_edges_summary_post --kwargs '{"v1_edges_summary_post_request": {"filter": {}}}' graphiant invoke v1_global_summary_post --kwargs '{"v1_global_summary_post_request": {"ntpType": true}}' + graphiant invoke v1_global_content_filters_get + graphiant invoke v1_global_domain_categories_get ``` - **`graphiant rest`** — Query strings are a single **`--query`** / **`-q`** string: **`key=value`** pairs joined with **`&`**. Values are strings (URL-encode special characters in the shell if needed). diff --git a/graphiant_api_docs_v26.4.1.json b/graphiant_api_docs_v26.4.0.json similarity index 100% rename from graphiant_api_docs_v26.4.1.json rename to graphiant_api_docs_v26.4.0.json diff --git a/graphiant_cli/sdk_invoke.py b/graphiant_cli/sdk_invoke.py index 79563ddd..19e661e6 100644 --- a/graphiant_cli/sdk_invoke.py +++ b/graphiant_cli/sdk_invoke.py @@ -5,9 +5,10 @@ import inspect import json import re +import types from functools import lru_cache from pathlib import Path -from typing import Annotated, Any, Optional, get_args, get_origin +from typing import Annotated, Any, Optional, Union, get_args, get_origin from pydantic import BaseModel @@ -71,10 +72,45 @@ def _unwrap_annotated(annotation: Any) -> Any: return annotation +def _is_union_origin(origin: Any) -> bool: + if origin is Union: + return True + ut = getattr(types, "UnionType", None) + return ut is not None and origin is ut + + def _coerce_value(value: Any, annotation: Any) -> Any: if annotation is None or annotation is inspect.Parameter.empty: return value inner = _unwrap_annotated(annotation) + origin = get_origin(inner) + args = get_args(inner) + + if _is_union_origin(origin): + non_none = [a for a in args if a is not type(None)] + if value is None: + return None + if len(non_none) == 1: + return _coerce_value(value, non_none[0]) + return value + + if origin is list: + if not isinstance(value, list) or len(args) != 1: + return value + elt_ann = _unwrap_annotated(args[0]) + elt_origin = get_origin(elt_ann) + elt_args = get_args(elt_ann) + if _is_union_origin(elt_origin): + nn = [a for a in elt_args if a is not type(None)] + if len(nn) == 1: + elt_ann = nn[0] + if isinstance(elt_ann, type) and issubclass(elt_ann, BaseModel): + return [ + elt_ann.model_validate(item) if isinstance(item, dict) else item + for item in value + ] + return value + if isinstance(value, dict) and isinstance(inner, type) and issubclass(inner, BaseModel): return inner.model_validate(value) return value diff --git a/pyproject.toml b/pyproject.toml index 345744c7..1e971be9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,6 +44,10 @@ filelock = ">=3.20.3" requires = ["setuptools"] build-backend = "setuptools.build_meta" +[tool.pytest.ini_options] +pythonpath = ["."] +testpaths = ["tests"] + [tool.pylint.'MESSAGES CONTROL'] extension-pkg-whitelist = "pydantic" diff --git a/tests/test_sdk_invoke.py b/tests/test_sdk_invoke.py index 3aeb0284..2a4d6143 100644 --- a/tests/test_sdk_invoke.py +++ b/tests/test_sdk_invoke.py @@ -1,6 +1,11 @@ """Tests for SDK method listing (no live API calls).""" -from graphiant_cli.sdk_invoke import list_api_method_rows, list_api_methods +from typing import Optional + +from graphiant_cli.sdk_invoke import _coerce_value, list_api_method_rows, list_api_methods +from graphiant_sdk.models.v1_global_content_filters_post_request import ( + V1GlobalContentFiltersPostRequest, +) def test_list_api_methods_includes_auth_get() -> None: @@ -18,3 +23,36 @@ def test_list_api_method_rows_includes_raw_http() -> None: rows = {name: (verb, path) for name, verb, path in list_api_method_rows("v1_edges_summary")} assert rows["v1_edges_summary_get"] == ("GET", "/v1/edges-summary") assert rows["v1_edges_summary_post"][0] == "POST" + + +def test_list_api_methods_includes_global_content_filters() -> None: + names = list_api_methods("v1_global_content_filters") + assert "v1_global_content_filters_get" in names + assert "v1_global_content_filters_post" in names + + +def test_list_api_method_rows_global_content_filters_and_rollouts() -> None: + rows = {n: (v, p) for n, v, p in list_api_method_rows("v1_global_content_filters")} + assert rows["v1_global_content_filters_get"] == ("GET", "/v1/global/content-filters") + assert rows["v1_global_content_filters_post"][0] == "POST" + roll = {n: (v, p) for n, v, p in list_api_method_rows("v1_software_rollouts")} + assert roll["v1_software_rollouts_get"] == ("GET", "/v1/software/rollouts") + macsec = {n: (v, p) for n, v, p in list_api_method_rows("v2_monitoring_macsec_device_id_status")} + assert macsec["v2_monitoring_macsec_device_id_status_get"] == ( + "GET", + "/v2/monitoring/macsec/{deviceId}/status", + ) + + +def test_coerce_optional_request_model_from_dict() -> None: + """CLI --kwargs JSON dicts become Pydantic bodies (incl. Optional[Model] params).""" + ann = Optional[V1GlobalContentFiltersPostRequest] + out = _coerce_value({}, ann) + assert isinstance(out, V1GlobalContentFiltersPostRequest) + out2 = _coerce_value(None, ann) + assert out2 is None + + +def test_coerce_request_model_non_optional() -> None: + out = _coerce_value({}, V1GlobalContentFiltersPostRequest) + assert isinstance(out, V1GlobalContentFiltersPostRequest)