jerry/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-10-27 04:31:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'v5' of https://github.com/OpenBB-finance/OpenBB into v5

Browse Source
This commit is contained in:
Danglewood
2026-09-25 14:32:50 -07:00
parent 32b6d4a280 6e138eee4c
commit 940c9b4804
12 changed files with 1241 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
cli/conftest.py
View File

@@ -275,11 +275,15 @@ def run_in_obb(cli_fake_extension_installed, tmp_path_factory): # noqa: ARG001
text=True,
env=_subprocess_env({"OPENBB_AUTO_BUILD": "false"}),
)
if worker.stdin is None or worker.stdout is None:
raise RuntimeError("obb worker started without piped stdin/stdout")
worker_stdin = worker.stdin
worker_stdout = worker.stdout
def _run(snippet: str) -> dict:
worker.stdin.write(json.dumps(textwrap.dedent(snippet).strip()) + "\n")
worker.stdin.flush()
line = worker.stdout.readline()
worker_stdin.write(json.dumps(textwrap.dedent(snippet).strip()) + "\n")
worker_stdin.flush()
line = worker_stdout.readline()
if not line:
raise AssertionError(
f"obb worker exited (rc={worker.wait()})\n"
@@ -293,5 +297,5 @@ def run_in_obb(cli_fake_extension_installed, tmp_path_factory): # noqa: ARG001
try:
yield _run
finally:
worker.stdin.close()
worker_stdin.close()
worker.wait(timeout=60)

20
cli/openbb_cli/codegen/credentials.py
View File

@@ -4,7 +4,7 @@ from __future__ import annotations
from typing import Any, Literal
CredentialLocation = Literal["query", "header"]
CredentialLocation = Literal["query", "header", "cookie"]
_CREDENTIAL_NAMES: frozenset[str] = frozenset(
{
@@ -50,12 +50,24 @@ def is_credential_name(name: str) -> bool:
def classify_parameter(param: dict[str, Any]) -> CredentialLocation | None:
"""Return ``"query"`` / ``"header"`` for a credential param, ``None`` otherwise."""
"""Return ``"query"`` / ``"header"`` / ``"cookie"`` for a credential param.
A parameter is a credential when its name is a known secret alias
(``api_key``, ``token``, ...), or when it's a *required* ``header``/
``cookie`` parameter regardless of name — that's the profile of a
per-integration value (a tenant id, a session cookie) declared once on a
path item for every operation under it, not something request-specific
like ``If-Modified-Since``. Returns ``None`` for anything else.
"""
name = param.get("name") if isinstance(param, dict) else None
if not isinstance(name, str) or not is_credential_name(name):
if not isinstance(name, str) or not name:
return None
location = param.get("in", "query")
return "header" if location == "header" else "query"
if is_credential_name(name):
return "header" if location == "header" else "query"
if param.get("required") and location in ("header", "cookie"):
return location
return None
def credentials_from_command(

48
cli/openbb_cli/codegen/fetcher_gen.py
View File

@@ -66,7 +66,7 @@ class GeneratedFetcher:
source : str
Full module source ready to write to disk.
credentials_used : dict
``{canonical_name: {"name": wire_name, "in": "query"|"header"}}``.
``{canonical_name: {"name": wire_name, "in": "query"|"header"|"cookie"}}``.
"""
module_name: str
@@ -491,6 +491,30 @@ def _header_dict_construction(creds: dict[str, dict[str, str]]) -> str:
return "\n".join(lines)
def _cookie_dict_construction(creds: dict[str, dict[str, str]]) -> str:
"""Compose the ``_cookies = {...}`` block for credential cookies.
Parameters
----------
creds : dict
Credential entries.
Returns
-------
str
Python source for the cookies dict.
"""
lines = [" _cookies: dict[str, str] = {}"]
for canonical, info in creds.items():
if info["in"] != "cookie":
continue
var = f"_cred_{canonical}"
wire_name = info["name"]
lines.append(f" if {var}:")
lines.append(f" _cookies[{wire_name!r}] = {var}")
return "\n".join(lines)
def _resolve_url_path(api_prefix: str, url_path: str) -> str:
"""Combine api_prefix with url_path; the prefix is dropped if already present."""
prefix = ("/" + api_prefix.strip("/")) if api_prefix.strip("/") else ""
@@ -568,6 +592,7 @@ def generate_fetcher_module(spec: FetcherCommandSpec) -> GeneratedFetcher:
spec.cmd_spec, creds, path_params, spec.provider_name
)
header_block = _header_dict_construction(creds)
cookie_block = _cookie_dict_construction(creds)
if _command_is_streaming(spec.cmd_spec):
source = _render_streaming_fetcher_source(
@@ -582,6 +607,7 @@ def generate_fetcher_module(spec: FetcherCommandSpec) -> GeneratedFetcher:
cred_lines=cred_lines,
query_block=query_block,
header_block=header_block,
cookie_block=cookie_block,
description=description,
)
return GeneratedFetcher(
@@ -610,6 +636,7 @@ def generate_fetcher_module(spec: FetcherCommandSpec) -> GeneratedFetcher:
cred_lines=cred_lines,
query_block=query_block,
header_block=header_block,
cookie_block=cookie_block,
creds=creds,
description=description,
imports=consolidate_imports(imports),
@@ -640,6 +667,7 @@ def _render_streaming_fetcher_source(
cred_lines: list[str],
query_block: str,
header_block: str,
cookie_block: str,
description: str,
) -> str:
"""Format a streaming fetcher module source.
@@ -668,6 +696,8 @@ def _render_streaming_fetcher_source(
Pre-rendered query-dict source.
header_block : str
Pre-rendered headers-dict source.
cookie_block : str
Pre-rendered cookies-dict source.
description : str
First paragraph of the spec command's description.
@@ -749,6 +779,7 @@ def _render_streaming_fetcher_source(
' + ("?" + _query_string if _query_string else "")'
)
parts.append(header_block)
parts.append(cookie_block)
parts.append("")
parts.append(" async def _stream() -> AsyncIterator[str]:")
parts.append(
@@ -756,7 +787,7 @@ def _render_streaming_fetcher_source(
)
parts.append(
f' async with _client.stream("{method.upper()}", '
"_url, headers=_headers) as _resp:"
"_url, headers=_headers, cookies=_cookies) as _resp:"
)
parts.append(" _resp.raise_for_status()")
parts.append(" async for _line in _resp.aiter_lines():")
@@ -792,6 +823,7 @@ def _render_fetcher_source(
cred_lines: list[str],
query_block: str,
header_block: str,
cookie_block: str,
creds: dict[str, dict[str, str]],
description: str,
imports: list[str],
@@ -819,8 +851,8 @@ def _render_fetcher_source(
Names of placeholders in ``url_path_template``.
cred_lines : list of str
Credential-extraction source lines.
query_block, header_block : str
Pre-rendered query-dict and headers-dict source.
query_block, header_block, cookie_block : str
Pre-rendered query-dict, headers-dict, and cookies-dict source.
creds : dict
Credential entries.
description : str
@@ -926,6 +958,7 @@ def _render_fetcher_source(
cred_lines=cred_lines,
query_block=query_block,
header_block=header_block,
cookie_block=cookie_block,
body_field_names=list(
(spec.cmd_spec.get("request_body_schema") or {}).get("properties") or {}
),
@@ -993,6 +1026,7 @@ def _aextract_body(
cred_lines: list[str],
query_block: str,
header_block: str,
cookie_block: str,
body_field_names: list[str],
has_path_params: bool,
has_query_creds: bool,
@@ -1015,6 +1049,8 @@ def _aextract_body(
Pre-rendered query-dict construction block.
header_block : str
Pre-rendered headers-dict construction block.
cookie_block : str
Pre-rendered cookies-dict construction block.
body_field_names : list of str
Names of fields that should travel in the JSON request body.
has_path_params : bool
@@ -1042,6 +1078,8 @@ def _aextract_body(
lines.append("")
lines.append(header_block)
lines.append("")
lines.append(cookie_block)
lines.append("")
if method == "post":
body_literal = (
"{"
@@ -1060,7 +1098,7 @@ def _aextract_body(
lines.append(" async with await get_async_requests_session() as _session:")
lines.append(
" async with await _session.request("
"_method, _url, headers=_headers, **_request_kwargs) as _resp:"
"_method, _url, headers=_headers, cookies=_cookies, **_request_kwargs) as _resp:"
)
lines.append(
" _ct = (_resp.headers.get('Content-Type') or '').lower()"

74
cli/openbb_cli/codegen/post_gen.py
View File

@@ -357,11 +357,35 @@ def _signature_params(
body_required = set(
(cmd_spec.get("request_body_schema") or {}).get("required") or []
)
# A name can be declared both as an operation parameter and as a request-body
# property — Codat's push endpoints put ``accountId`` in the URL template and
# in the body. Emitting both produces two parameters with the same name, which
# is a SyntaxError. The operation parameter wins: it carries the path
# placeholder and its own ``required``. The payload is unaffected, because
# ``_render_body_block`` reads the body fields from ``request_body_schema``
# rather than from this list.
#
# Two wire names can also collapse onto one identifier once sanitized
# (``from`` and ``from_``, ``Organization-Id`` and ``Organization_Id``), and
# ``operation_parameters`` keeps a name declared twice under different ``in``
# locations. Collisions are therefore resolved on the emitted identifier, not
# on the wire name, so a duplicate argument cannot be generated at all.
operation_params = [
p for p in filter_user_params(cmd_spec.get("parameters") or []) if p.get("name")
]
operation_param_names = {p["name"] for p in operation_params}
operation_param_names |= {safe_field_name(p["name"])[0] for p in operation_params}
emitted: set[str] = {entry[0] for entry in out}
for name, schema in body_props.items():
if name == array_field:
continue
if not isinstance(schema, dict):
continue
safe = safe_field_name(name)[0]
if safe in operation_param_names or safe in emitted:
continue
emitted.add(safe)
ann = _python_type_from_param(
{
"type": schema.get("type"),
@@ -371,21 +395,31 @@ def _signature_params(
)
out.append(
(
name,
# ``_render_body_block`` already writes ``{'from': from_}``, keying the
# payload by the wire name and reading the safe identifier, so the
# signature has to declare the safe one. Codat's Transfer body has a
# property named ``from``, which is a keyword.
safe,
ann,
schema.get("description") or schema.get("title"),
name in body_required,
schema.get("default"),
)
)
for raw in filter_user_params(cmd_spec.get("parameters") or []):
name = raw.get("name")
if not name:
for raw in operation_params:
name = raw["name"]
safe = safe_field_name(name)[0]
if safe in emitted:
continue
emitted.add(safe)
ann = _python_type_from_param(raw)
out.append(
(
name,
# Wire names are not always identifiers. Rebilly declares
# ``Organization-Id`` and a ``REB-APIKEY`` security header, which
# emitted ``REB-APIKEY: str = None``. The fetcher path already
# sanitizes these; this one has to match it.
safe,
ann,
raw.get("help"),
bool(raw.get("required")),
@@ -625,9 +659,21 @@ def _render_body_block(
"""
body_props = (cmd_spec.get("request_body_schema") or {}).get("properties") or {}
body_field_names = list(body_props)
# The signature declares the safe identifier for a body property, so the
# body/query split has to compare against those, not the wire names.
# Otherwise a keyword-named property such as ``from`` is not recognized as a
# body field and leaks into the query string as ``from_``.
body_safe_names = {safe_field_name(n)[0] for n in body_field_names}
query_field_names = [
n for n, _, _, _, _ in params if n != "cc" and n not in body_field_names
n for n, _, _, _, _ in params if n != "cc" and n not in body_safe_names
]
# The signature carries safe identifiers, but the request has to go out under
# the wire names, so keep a map back for the query keys and the path check.
wire_by_safe = {
safe_field_name(p["name"])[0]: p["name"]
for p in filter_user_params(cmd_spec.get("parameters") or [])
if p.get("name")
}
lines: list[str] = []
lines.extend(cred_lines)
@@ -640,10 +686,11 @@ def _render_body_block(
lines.append(" _query_dict: dict[str, Any] = {}")
for name in query_field_names:
if name in path_params:
wire = wire_by_safe.get(name, name)
if wire in path_params or name in path_params:
continue
lines.append(f" if {name} is not None:")
lines.append(f" _query_dict[{name!r}] = {name}")
lines.append(f" _query_dict[{wire!r}] = {name}")
for canonical, info in creds.items():
if info["in"] != "query":
continue
@@ -665,6 +712,15 @@ def _render_body_block(
lines.append(f" if {var}:")
lines.append(f" _headers[{wire_name!r}] = {var}")
lines.append(" _cookies: dict[str, str] = {}")
for canonical, info in creds.items():
if info["in"] != "cookie":
continue
wire_name = info["name"]
var = f"_cred_{canonical}"
lines.append(f" if {var}:")
lines.append(f" _cookies[{wire_name!r}] = {var}")
body_top_is_array = (cmd_spec.get("request_body_schema") or {}).get(
"type"
) == "array"
@@ -695,7 +751,7 @@ def _render_body_block(
lines.append(" async with await get_async_requests_session() as _session:")
lines.append(
" async with await _session.request("
'"POST", _url, headers=_headers, json=_body) as _resp:'
'"POST", _url, headers=_headers, cookies=_cookies, json=_body) as _resp:'
)
lines.append(" _ct = (_resp.headers.get('Content-Type') or '').lower()")
lines.append(" _text = await _resp.text()")

5
cli/openbb_cli/codegen/test_gen.py
View File

@@ -182,6 +182,11 @@ def _merge_redactions(
continue
if location == "header":
headers[wire] = _mock_value_for(wire)
elif location == "cookie":
# Cookies travel on the wire as the ``Cookie`` header, which
# ``_STANDARD_HEADER_REDACTIONS`` already blanket-redacts below
# — VCR has no way to filter a single cookie within it.
continue
else:
queries[wire] = _mock_value_for(wire)
for n, v in _STANDARD_HEADER_REDACTIONS:

215
cli/openbb_cli/dispatchers/openapi_schema.py
View File

@@ -10,6 +10,8 @@ from urllib.parse import unquote, urldefrag, urljoin, urlsplit
import httpx
_INHERITABLE_PARAMETER_LOCATIONS = frozenset({"path", "query"})
PROVIDER_TAG_RE = re.compile(r"\s*\(provider:\s*([^)]+)\)\s*$")
PROVIDER_SECTION_SPLIT_RE = re.compile(r";\s*\n\s*")
@@ -122,6 +124,79 @@ def deref_parameter(spec: dict[str, Any], param: dict[str, Any]) -> dict[str, An
return param
def operation_parameters(
spec: dict[str, Any], path_item: dict[str, Any], op: dict[str, Any]
) -> list[dict[str, Any]]:
"""Return an operation's parameters, including those inherited from its path item.
OpenAPI 3.x lets a Path Item Object declare ``parameters`` that apply to every
operation under that path. An operation-level parameter overrides an inherited
one when both match on ``(name, in)``; anything else is inherited as-is.
Only ``in: path`` and ``in: query`` are inherited. A path-item ``in: header``
or ``in: cookie`` entry is deliberately dropped: those do not translate into a
Python interface function, and outbound headers belong to transport
configuration (``SystemSettings.PythonSettings.http.headers``, the CLI's
``-H``) or to security-scheme handling — not to a per-command argument.
Entries are returned dereferenced, inherited first, then the operation's own.
"""
inherited = path_item.get("parameters") or []
own_raw = op.get("parameters") or []
own = [deref_parameter(spec, p) for p in own_raw if isinstance(p, dict)]
own = [p for p in own if p]
if not inherited:
return own
overridden = {(p.get("name"), p.get("in")) for p in own}
merged: list[dict[str, Any]] = []
for raw in inherited:
if not isinstance(raw, dict):
continue
resolved = deref_parameter(spec, raw)
if not resolved or not resolved.get("name"):
continue
if resolved.get("in") not in _INHERITABLE_PARAMETER_LOCATIONS:
continue
if (resolved.get("name"), resolved.get("in")) in overridden:
continue
merged.append(resolved)
merged.extend(own)
return merged
def path_item_credential_parameters(
spec: dict[str, Any], path_item: dict[str, Any]
) -> list[dict[str, Any]]:
"""Return the path item's required ``header``/``cookie`` parameters, dereferenced.
These are exactly what :func:`operation_parameters` drops: a value every
operation under the path needs (a tenant id, a session cookie) that isn't a
per-call argument. They're a per-integration secret, not different per
request, so they route through ``openbb_cli.codegen.credentials`` the same
way an ``apiKey`` security scheme does, rather than becoming a required
argument the caller has to supply on every command.
A path-item parameter that is optional is left for the caller to notice —
only a required one is dead-on-arrival without this.
"""
out: list[dict[str, Any]] = []
for raw in path_item.get("parameters") or []:
if not isinstance(raw, dict):
continue
resolved = deref_parameter(spec, raw)
if not resolved or not resolved.get("name"):
continue
if resolved.get("in") not in ("header", "cookie"):
continue
if not resolved.get("required"):
continue
out.append(resolved)
return out
_NON_SCHEMA_KEYWORDS = frozenset({"default", "example", "examples", "enum", "const"})
def deref_schema(
spec: dict[str, Any],
node: Any,
@@ -142,12 +217,125 @@ def deref_schema(
if not target:
return node
return deref_schema(spec, target, seen | {ref}, max_depth - 1)
return {k: deref_schema(spec, v, seen, max_depth - 1) for k, v in node.items()}
# Instance data (``default``, ``example(s)``, ``enum``, ``const``) and
# vendor extensions are not subschemas: a ``$ref`` key inside one is a
# literal value, not a reference to resolve.
return {
k: (
v
if k in _NON_SCHEMA_KEYWORDS or k.startswith("x-")
else deref_schema(spec, v, seen, max_depth - 1)
)
for k, v in node.items()
}
if isinstance(node, list):
return [deref_schema(spec, v, seen, max_depth - 1) for v in node]
return node
_ALLOF_BLOCKING_KEYS = frozenset(
{"oneOf", "anyOf", "allOf", "not", "$ref", "enum", "const", "items"}
)
def _mergeable_allof_members(members: list[Any]) -> list[dict[str, Any]] | None:
"""Return the members if every one is a plain object subschema, else ``None``.
A member is not mergeable when it carries its own combinator (``oneOf`` /
``anyOf`` / a nested ``allOf`` that did not collapse), an unresolved
``$ref`` left behind by a reference cycle, a scalar constraint (``enum`` /
``const`` / ``items``), or a ``type`` other than ``object``. Any of those
means the composition is not a plain intersection of property bags, and
merging it would describe something the spec does not.
Everything else is annotation that does not affect the intersection —
``title``, ``description``, ``examples``, ``definitions``, vendor ``x-``
keys — and does not block the merge. Keying off a blocklist rather than a
whitelist matters: Codat's ``PagingInfo`` carries ``definitions``, and
treating that as disqualifying left the composition unmerged.
"""
out: list[dict[str, Any]] = []
for member in members:
if not isinstance(member, dict):
return None
if member.keys() & _ALLOF_BLOCKING_KEYS:
return None
declared = member.get("type")
if declared is not None and declared != "object":
return None
out.append(member)
return out
def merge_allof(node: Any, max_depth: int = 32) -> Any:
"""Collapse ``allOf`` compositions of object subschemas into one object schema.
An ``allOf`` is an intersection: the instance must satisfy every member. A
consumer reading ``properties`` off the composition itself finds nothing,
because the properties live one level down inside the members. This merges
them into a single object schema so the composition can be read like any
other object.
``properties`` are unioned, with the first member to declare a name winning
on conflict, and any sibling ``properties`` on the ``allOf`` node itself
taking precedence over all members. ``required`` is unioned. Compositions
that are not a plain intersection of object subschemas are returned
unchanged.
"""
if max_depth <= 0:
return node
if isinstance(node, list):
return [merge_allof(v, max_depth - 1) for v in node]
if not isinstance(node, dict):
return node
# ``default``, ``example(s)``, ``enum`` and ``const`` hold instance data, not
# subschemas, and a vendor extension can hold anything. Recursing into them
# would rewrite a value that merely happens to contain an ``allOf`` key —
# a default of ``{"allOf": [...]}`` came back as ``{"type": "object"}``.
node = {
k: (
v
if k in _NON_SCHEMA_KEYWORDS or k.startswith("x-")
else merge_allof(v, max_depth - 1)
)
for k, v in node.items()
}
members = node.get("allOf")
if not isinstance(members, list) or not members:
return node
mergeable = _mergeable_allof_members(members)
if mergeable is None:
return node
properties: dict[str, Any] = {}
required: list[str] = []
for member in mergeable:
for name, subschema in (member.get("properties") or {}).items():
properties.setdefault(name, subschema)
for name in member.get("required") or []:
if name not in required:
required.append(name)
siblings = {k: v for k, v in node.items() if k != "allOf"}
# Keywords written next to the ``allOf`` describe the composition itself and
# outrank anything the members declare.
for name, subschema in (siblings.get("properties") or {}).items():
properties[name] = subschema
for name in siblings.get("required") or []:
if name not in required:
required.append(name)
merged: dict[str, Any] = {**siblings, "type": "object"}
if properties:
merged["properties"] = properties
if required:
merged["required"] = required
return merged
_SUCCESS_PRIORITY = ("200", "2XX", "201", "default")
_JSON_CONTENT_TYPES = ("application/json", "application/vnd.api+json")
@@ -193,7 +381,7 @@ def extract_response_schema(
schema = media.get("schema")
if not isinstance(schema, dict):
return None
return deref_schema(spec, schema)
return merge_allof(deref_schema(spec, schema))
def extract_request_body_schema(
@@ -222,7 +410,7 @@ def extract_request_body_schema(
schema = media.get("schema")
if not isinstance(schema, dict):
return None
return deref_schema(spec, schema)
return merge_allof(deref_schema(spec, schema))
def extract_response_schemas(
@@ -242,7 +430,7 @@ def extract_response_schemas(
continue
schema = media.get("schema")
if isinstance(schema, dict):
per_content[content_type] = deref_schema(spec, schema)
per_content[content_type] = merge_allof(deref_schema(spec, schema))
if per_content:
out[status] = per_content
return out
@@ -426,9 +614,15 @@ def parameter_to_kwargs(param: dict[str, Any]) -> tuple[str, dict[str, Any]] | N
def build_parser_from_operation(
op: dict[str, Any], spec: dict[str, Any] | None = None
op: dict[str, Any],
spec: dict[str, Any] | None = None,
path_item: dict[str, Any] | None = None,
) -> argparse.ArgumentParser:
"""Build an ``ArgumentParser`` from an OpenAPI operation object."""
"""Build an ``ArgumentParser`` from an OpenAPI operation object.
When ``path_item`` is given, parameters shared by every operation under that
path are inherited by the parser (see :func:`operation_parameters`).
"""
parser = argparse.ArgumentParser(
prog=op.get("operationId", "cmd"),
description=(op.get("description") or op.get("summary") or "").strip() or None,
@@ -440,7 +634,12 @@ def build_parser_from_operation(
if spec is not None
else []
)
for param in [*op.get("parameters", []), *body_params]:
params = (
operation_parameters(spec, path_item or {}, op)
if spec is not None
else list(op.get("parameters", []) or [])
)
for param in [*params, *body_params]:
translated = parameter_to_kwargs(param)
if translated is None:
continue
@@ -509,7 +708,7 @@ def build_command_index(
if not op:
continue
index[url_to_command(url, api_prefix=api_prefix)] = build_parser_from_operation(
op, spec
op, spec, methods
)
return index

24
cli/openbb_cli/dispatchers/spec.py
View File

@@ -22,13 +22,14 @@ from openbb_cli.dispatchers.openapi_schema import (
_resolve_schema,
build_reference,
build_router_map,
deref_parameter,
detect_api_prefix,
extract_request_body_schema,
extract_response_schema,
extract_response_schemas,
operation_parameters,
param_provider_membership,
parse_json_arg,
path_item_credential_parameters,
request_body_parameters,
url_to_command,
)
@@ -190,20 +191,33 @@ def _security_parameters(
def _build_operation_entry(
spec: dict[str, Any], url: str, method: str, op: dict[str, Any]
spec: dict[str, Any],
url: str,
method: str,
op: dict[str, Any],
path_item: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Build the per-URL command entry — params, providers, schemas."""
providers = _operation_providers(op)
providers_set = set(providers) if providers else None
params: list[dict[str, Any]] = []
for raw in op.get("parameters", []) or []:
resolved = deref_parameter(spec, raw) if isinstance(raw, dict) else raw
for resolved in operation_parameters(spec, path_item or {}, op):
if not resolved:
continue
normalized = _normalize_parameter(resolved, providers_set)
if normalized is not None:
params.append(normalized)
declared = {p["name"] for p in params}
# A required header/cookie the path item declares for every operation (a
# tenant id, a session cookie) isn't a per-call argument — it routes through
# credentials.classify_parameter, same as an apiKey security scheme.
for resolved in path_item_credential_parameters(spec, path_item or {}):
if resolved["name"] in declared:
continue
normalized = _normalize_parameter(resolved, providers_set)
if normalized is not None:
params.append(normalized)
declared.add(normalized["name"])
for raw in _security_parameters(spec, op):
if raw["name"] in declared:
continue
@@ -298,7 +312,7 @@ def build_command_spec(
if not base_cmd:
continue
groups.setdefault(base_cmd, []).append(
_build_operation_entry(spec, url, method, op)
_build_operation_entry(spec, url, method, op, methods)
)
out: dict[str, dict[str, Any]] = {}
for cmd, entries in groups.items():

46
cli/tests/test_codegen_credentials.py
View File

@@ -104,6 +104,38 @@ def test_classify_parameter_returns_none_for_non_dict_input():
assert cr.classify_parameter("not-a-dict") is None # type: ignore[arg-type]
def test_classify_parameter_returns_header_for_required_unnamed_header():
"""A required header credential is recognized by shape, not just by name.
e.g. Xero's ``xero-tenant-id`` — a required per-tenant header every
operation needs, but not a name on the known-secret allowlist.
"""
p = {"name": "xero-tenant-id", "in": "header", "required": True}
assert cr.classify_parameter(p) == "header"
def test_classify_parameter_returns_cookie_for_required_unnamed_cookie():
p = {"name": "xero-session", "in": "cookie", "required": True}
assert cr.classify_parameter(p) == "cookie"
def test_classify_parameter_returns_none_for_optional_unnamed_header():
"""An optional, non-secret-named header stays a regular parameter."""
p = {"name": "If-Modified-Since", "in": "header", "required": False}
assert cr.classify_parameter(p) is None
def test_classify_parameter_returns_none_for_optional_unnamed_cookie():
p = {"name": "session", "in": "cookie"}
assert cr.classify_parameter(p) is None
def test_classify_parameter_known_name_wins_over_required_check():
"""A known secret name still classifies query/header, even if required."""
p = {"name": "api_key", "in": "query", "required": True}
assert cr.classify_parameter(p) == "query"
# --- credentials_from_command ---
@@ -122,6 +154,20 @@ def test_credentials_from_command_extracts_query_and_header_entries():
}
def test_credentials_from_command_extracts_required_cookie_entry():
cmd = {
"parameters": [
{"name": "xero-tenant-id", "in": "header", "required": True},
{"name": "xero-session", "in": "cookie", "required": True},
]
}
out = cr.credentials_from_command(cmd)
assert out == {
"xero_tenant_id": {"name": "xero-tenant-id", "in": "header"},
"xero_session": {"name": "xero-session", "in": "cookie"},
}
def test_credentials_from_command_returns_empty_when_no_creds():
assert cr.credentials_from_command({"parameters": [{"name": "limit"}]}) == {}

66
cli/tests/test_codegen_fetcher_gen.py
View File

@@ -843,6 +843,27 @@ def test_header_dict_construction_skips_query_credentials():
assert "apikey" not in out
# --- _cookie_dict_construction ---
def test_cookie_dict_construction_emits_cookie_credentials():
creds = {"xero_session": {"name": "xero-session", "in": "cookie"}}
out = fg._cookie_dict_construction(creds)
assert "_cookies: dict[str, str] = {}" in out
assert "if _cred_xero_session:" in out
assert "_cookies['xero-session'] = _cred_xero_session" in out
def test_cookie_dict_construction_skips_header_and_query_credentials():
creds = {
"authorization": {"name": "Authorization", "in": "header"},
"api_key": {"name": "apikey", "in": "query"},
}
out = fg._cookie_dict_construction(creds)
assert "Authorization" not in out
assert "apikey" not in out
# --- generate_fetcher_module (end-to-end) ---
@@ -1030,3 +1051,48 @@ def test_generate_fetcher_module_credentials_round_trip_to_query_string():
src = out.source
assert f'_cred_{canonical} = _creds.get("fmp_{canonical}", "")' in src
assert f"_query_dict['apikey'] = _cred_{canonical}" in src
def test_generate_fetcher_module_required_cookie_reaches_the_request():
"""A required path-item cookie (Xero's tenant-scoped session, e.g.) is sent.
Regression guard for the case ``operation_parameters`` intentionally never
fixes: a required header/cookie the path item declares for every operation
isn't a CLI/Python-interface argument, it's a stored credential — but it
still has to actually reach the outgoing request.
"""
spec = fg.FetcherCommandSpec(
name="accounting.accounts",
cmd_spec={
"parameters": [
{
"name": "xero-tenant-id",
"type": "string",
"in": "header",
"required": True,
},
{
"name": "xero-session",
"type": "string",
"in": "cookie",
"required": True,
},
],
"url_path": "/Accounts",
"method": "get",
"description": "List accounts.",
"response_schema": {"type": "object"},
},
base_url="https://api.xero.com/api.xro/2.0",
api_prefix="",
provider_name="xero",
)
out = fg.generate_fetcher_module(spec)
assert set(out.credentials_used) == {"xero_tenant_id", "xero_session"}
src = out.source
assert "_cred_xero_tenant_id" in src
assert "_cred_xero_session" in src
assert "_headers['xero-tenant-id'] = _cred_xero_tenant_id" in src
assert "_cookies['xero-session'] = _cred_xero_session" in src
assert "cookies=_cookies" in src
ast.parse(src)

163
cli/tests/test_codegen_post_gen.py
View File

@@ -660,6 +660,37 @@ def test_generate_post_command_module_with_credentials_renders_cred_block():
ast.parse(src)
def test_generate_post_command_module_with_cookie_credential_renders_cookie_block():
"""A required path-item cookie (e.g. a session id) reaches the request as a cookie."""
spec = pg.PostCommandSpec(
name="x.y",
cmd_spec={
"url_path": "/x",
"method": "post",
"description": "X.",
"parameters": [
{
"name": "xero-session",
"type": "string",
"in": "cookie",
"required": True,
},
],
"request_body_schema": {"type": "object", "properties": {}},
"response_schema": {"type": "object"},
},
base_url="https://api.example.com",
api_prefix="",
provider_name="xero",
)
out = pg.generate_post_command_module(spec)
src = out.source
assert "_cred_xero_session" in src
assert "_cookies['xero-session']" in src
assert "cookies=_cookies" in src
ast.parse(src)
def test_generate_post_command_module_multiline_description_renders_block_doc():
spec = pg.PostCommandSpec(
name="x.y",
@@ -799,6 +830,138 @@ def test_generate_post_command_module_no_body_props_uses_empty_body_dict():
assert "_body: dict[str, Any] = {}" in out.source
def test_signature_params_body_property_does_not_duplicate_operation_parameter():
"""A name declared both as an operation parameter and a body property is emitted once.
Codat's push endpoints put ``accountId`` in the URL template and repeat it in
the request body. Emitting both produced two parameters with the same name,
which is a SyntaxError in the generated module.
"""
cmd = {
"url_path": "/companies/{companyId}/push/bankAccounts/{accountId}/transactions",
"parameters": [
{"name": "companyId", "type": "string", "required": True},
{"name": "accountId", "type": "string", "required": True},
],
"request_body_schema": {
"type": "object",
"properties": {
"accountId": {"type": "string"},
"reference": {"type": "string"},
},
},
}
out = pg._signature_params(
cmd, array_field=None, array_annotation=None, has_credentials=False
)
names = [e[0] for e in out]
assert names.count("accountId") == 1
assert len(names) == len(set(names))
# The operation parameter wins: it carries the path placeholder and required.
accountid = next(e for e in out if e[0] == "accountId")
assert accountid[3] is True
# Body-only properties are unaffected.
assert "reference" in names
def _keyword_body_cmd() -> dict:
return {
"url_path": "/companies/{companyId}/push/transfers",
"parameters": [{"name": "companyId", "type": "string", "required": True}],
"request_body_schema": {
"type": "object",
"properties": {
"from": {"type": "string"},
"to": {"type": "string"},
},
},
}
def test_signature_params_uses_safe_identifier_for_keyword_body_property():
"""A body property named after a Python keyword cannot be emitted verbatim.
Codat's Transfer body has a ``from`` property; ``from: str = None`` in a
signature is a SyntaxError.
"""
out = pg._signature_params(
_keyword_body_cmd(),
array_field=None,
array_annotation=None,
has_credentials=False,
)
names = [e[0] for e in out]
assert "from_" in names
assert "from" not in names
assert "to" in names
def test_render_body_block_keys_keyword_property_by_wire_name():
"""The payload keeps ``from``; only the local identifier is renamed."""
cmd = _keyword_body_cmd()
params = pg._signature_params(
cmd, array_field=None, array_annotation=None, has_credentials=False
)
block = pg._render_body_block(
cmd_spec=cmd,
body_field=None,
creds={},
path_params=["companyId"],
url_path_template="/companies/{companyId}/push/transfers",
base_url="https://api.codat.io",
cred_lines=[],
data_class="PushTransfersData",
params=params,
)
assert "'from': from_" in block
# A body property must not also be sent as a query parameter.
assert "_query_dict['from_']" not in block
assert '_query_dict["from_"]' not in block
def test_signature_params_never_emits_a_duplicate_identifier():
"""Collisions are resolved on the emitted identifier, not the wire name.
Two wire names can sanitize onto one identifier, and operation_parameters
keeps a name declared twice under different ``in`` locations. Neither may
produce a duplicate argument.
"""
cmd = {
"parameters": [
# same name, two locations — operation_parameters keeps both
{"name": "id", "in": "path", "type": "string", "required": True},
{"name": "id", "in": "query", "type": "string"},
# sanitizes onto the same identifier as the body's ``from``
{"name": "from_", "in": "query", "type": "string"},
{"name": "Organization-Id", "in": "header", "type": "string"},
],
"request_body_schema": {
"type": "object",
"properties": {
"from": {"type": "string"},
"Organization_Id": {"type": "string"},
"kept": {"type": "string"},
},
},
}
out = pg._signature_params(
cmd, array_field=None, array_annotation=None, has_credentials=True
)
names = [e[0] for e in out]
assert len(names) == len(set(names)), names
assert names.count("id") == 1
assert names.count("from_") == 1
assert names.count("Organization_Id") == 1
# A body property that collides with nothing is still emitted.
assert "kept" in names
# Every emitted name is a usable Python identifier.
assert all(n.isidentifier() for n in names)
def test_envelope_inner_schema_single_object_array_among_other_properties():
row = {"type": "object", "properties": {"symbol": {"type": "string"}}}
props = {

419
cli/tests/test_dispatchers_openapi_schema.py
View File

@@ -21,9 +21,15 @@ from openbb_cli.dispatchers.openapi_schema import (
build_parser_from_operation,
build_reference,
build_router_map,
deref_schema,
expand_type_arrays,
extract_request_body_schema,
extract_response_schema,
merge_allof,
operation_parameters,
parameter_to_kwargs,
parse_json_arg,
path_item_credential_parameters,
request_body_parameters,
url_to_command,
)
@@ -1875,6 +1881,419 @@ def test_bundle_external_refs_rejects_schema_ref_siblings(monkeypatch):
)
def test_operation_parameters_returns_own_when_no_path_item_params():
"""Without inherited parameters the operation's own list is returned, dereferenced."""
op = {"parameters": [{"name": "symbol", "in": "query"}]}
assert operation_parameters({}, {}, op) == [{"name": "symbol", "in": "query"}]
def test_operation_parameters_inherits_from_path_item():
"""Path-item parameters come first, then the operation's own."""
path_item = {"parameters": [{"name": "record_id", "in": "path"}]}
op = {"parameters": [{"name": "symbol", "in": "query"}]}
assert [p["name"] for p in operation_parameters({}, path_item, op)] == [
"record_id",
"symbol",
]
def test_operation_parameters_drops_inherited_headers_and_cookies():
"""Headers and cookies are transport concerns, not command arguments."""
path_item = {
"parameters": [
{"name": "x-tenant-id", "in": "header"},
{"name": "session", "in": "cookie"},
{"name": "record_id", "in": "path"},
]
}
assert [p["name"] for p in operation_parameters({}, path_item, {})] == ["record_id"]
def test_operation_parameters_keeps_operation_level_headers():
"""An operation's own header parameter is untouched by the inheritance filter."""
op = {"parameters": [{"name": "x-request-id", "in": "header"}]}
path_item = {"parameters": [{"name": "record_id", "in": "path"}]}
assert [p["name"] for p in operation_parameters({}, path_item, op)] == [
"record_id",
"x-request-id",
]
def test_operation_parameters_operation_overrides_inherited():
"""An operation parameter replaces an inherited one matching (name, in)."""
path_item = {"parameters": [{"name": "limit", "in": "query", "required": False}]}
op = {"parameters": [{"name": "limit", "in": "query", "required": True}]}
out = operation_parameters({}, path_item, op)
assert out == [{"name": "limit", "in": "query", "required": True}]
def test_operation_parameters_resolves_refs_on_both_levels():
"""``$ref`` entries are resolved before they are compared or returned."""
spec = {
"components": {
"parameters": {
"record_id": {"name": "record_id", "in": "path"},
"symbol": {"name": "symbol", "in": "query"},
}
}
}
path_item = {"parameters": [{"$ref": "#/components/parameters/record_id"}]}
op = {"parameters": [{"$ref": "#/components/parameters/symbol"}]}
assert [p["name"] for p in operation_parameters(spec, path_item, op)] == [
"record_id",
"symbol",
]
def test_operation_parameters_skips_unresolvable_and_unnamed_entries():
"""Unresolvable refs, non-dicts, and nameless entries are dropped."""
path_item = {
"parameters": [
{"$ref": "#/components/parameters/missing"},
"not-a-dict",
{"in": "query"},
{"name": "record_id", "in": "path"},
]
}
assert [p["name"] for p in operation_parameters({}, path_item, {})] == ["record_id"]
def test_path_item_credential_parameters_returns_required_header_and_cookie():
path_item = {
"parameters": [
{"name": "xero-tenant-id", "in": "header", "required": True},
{"name": "xero-session", "in": "cookie", "required": True},
]
}
out = path_item_credential_parameters({}, path_item)
assert [p["name"] for p in out] == ["xero-tenant-id", "xero-session"]
def test_path_item_credential_parameters_skips_optional_header_or_cookie():
"""An optional path-item header/cookie has no clear destination; it's dropped."""
path_item = {
"parameters": [
{"name": "If-Modified-Since", "in": "header"},
{"name": "session", "in": "cookie", "required": False},
]
}
assert path_item_credential_parameters({}, path_item) == []
def test_path_item_credential_parameters_ignores_path_and_query():
"""Path/query params are ``operation_parameters``'s job, not this one's."""
path_item = {
"parameters": [
{"name": "record_id", "in": "path", "required": True},
{"name": "page", "in": "query", "required": True},
]
}
assert path_item_credential_parameters({}, path_item) == []
def test_path_item_credential_parameters_resolves_refs():
spec = {
"components": {
"parameters": {
"tenant": {
"name": "xero-tenant-id",
"in": "header",
"required": True,
}
}
}
}
path_item = {"parameters": [{"$ref": "#/components/parameters/tenant"}]}
out = path_item_credential_parameters(spec, path_item)
assert [p["name"] for p in out] == ["xero-tenant-id"]
def test_path_item_credential_parameters_skips_malformed_entries():
path_item = {
"parameters": [
"not-a-dict",
{"$ref": "#/components/parameters/missing"},
{"in": "header", "required": True},
]
}
assert path_item_credential_parameters({}, path_item) == []
def test_build_command_index_inherits_path_item_parameters():
"""The interactive parser also picks up parameters shared by the path item."""
spec = {
"paths": {
"/api/v1/x": {
"parameters": [
{"name": "record_id", "in": "path", "schema": {"type": "string"}}
],
"get": {
"operationId": "x",
"parameters": [
{"name": "symbol", "in": "query", "schema": {"type": "string"}}
],
},
}
}
}
parser = build_command_index(spec)["x"]
dests = {a.dest for a in parser._actions}
assert "record_id" in dests
assert "symbol" in dests
def test_build_parser_from_operation_without_spec_ignores_path_item():
"""Called without a spec the parser keeps its previous operation-only behavior."""
op = {
"parameters": [{"name": "symbol", "in": "query", "schema": {"type": "string"}}]
}
parser = build_parser_from_operation(op)
assert {a.dest for a in parser._actions} == {"symbol"}
def test_merge_allof_unions_properties_and_required():
schema = {
"allOf": [
{"type": "object", "properties": {"results": {"type": "array"}}},
{
"type": "object",
"properties": {"pageNumber": {"type": "integer"}},
"required": ["pageNumber"],
},
]
}
merged = merge_allof(schema)
assert merged["type"] == "object"
assert set(merged["properties"]) == {"results", "pageNumber"}
assert merged["required"] == ["pageNumber"]
def test_merge_allof_first_member_wins_on_conflict():
schema = {
"allOf": [
{"properties": {"value": {"type": "string"}}},
{"properties": {"value": {"type": "integer"}}},
]
}
assert merge_allof(schema)["properties"]["value"] == {"type": "string"}
def test_merge_allof_sibling_keywords_outrank_members():
schema = {
"properties": {"value": {"type": "boolean"}},
"required": ["value"],
"allOf": [{"properties": {"value": {"type": "string"}}}],
}
merged = merge_allof(schema)
assert merged["properties"]["value"] == {"type": "boolean"}
assert merged["required"] == ["value"]
def test_merge_allof_nested_inside_properties():
schema = {
"type": "object",
"properties": {
"row": {
"allOf": [
{"properties": {"a": {"type": "string"}}},
{"properties": {"b": {"type": "string"}}},
]
}
},
}
merged = merge_allof(schema)
assert set(merged["properties"]["row"]["properties"]) == {"a", "b"}
def test_merge_allof_leaves_non_object_compositions_alone():
# A member carrying its own combinator, or a non-object type, is not a
# plain intersection of property bags and must not be flattened.
with_combinator = {
"allOf": [
{"properties": {"a": {"type": "string"}}},
{"oneOf": [{"type": "object"}, {"type": "null"}]},
]
}
scalar_member = {"allOf": [{"type": "string"}, {"properties": {"a": {}}}]}
assert merge_allof(with_combinator) == with_combinator
assert merge_allof(scalar_member) == scalar_member
def test_merge_allof_ignores_schemas_without_allof():
schema = {"type": "object", "properties": {"a": {"type": "string"}}}
assert merge_allof(schema) == schema
def test_extract_response_schema_merges_allof_composition():
spec = {
"components": {
"schemas": {
"Paged": {
"type": "object",
"properties": {"pageNumber": {"type": "integer"}},
"required": ["pageNumber"],
},
"Bills": {
"allOf": [
{
"type": "object",
"properties": {
"results": {
"type": "array",
"items": {"$ref": "#/components/schemas/Bill"},
}
},
},
{"$ref": "#/components/schemas/Paged"},
]
},
"Bill": {
"type": "object",
"properties": {"id": {"type": "string"}},
},
}
}
}
op = {
"responses": {
"200": {
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/Bills"}
}
}
}
}
}
schema = extract_response_schema(spec, op)
assert set(schema["properties"]) == {"results", "pageNumber"}
assert schema["required"] == ["pageNumber"]
# The nested reference inside the merged member is still resolved.
items = schema["properties"]["results"]["items"]
assert items["properties"]["id"] == {"type": "string"}
def test_extract_request_body_schema_merges_allof_composition():
spec = {
"components": {
"schemas": {
"Timestamps": {
"type": "object",
"properties": {"modifiedDate": {"type": "string"}},
},
"BillPrototype": {
"allOf": [
{
"type": "object",
"properties": {"reference": {"type": "string"}},
"required": ["reference"],
},
{"$ref": "#/components/schemas/Timestamps"},
]
},
}
}
}
op = {
"requestBody": {
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/BillPrototype"}
}
}
}
}
schema = extract_request_body_schema(spec, op)
assert set(schema["properties"]) == {"reference", "modifiedDate"}
assert schema["required"] == ["reference"]
# A composition body now flattens into body parameters instead of none.
names = {p["name"] for p in request_body_parameters(schema)}
assert names == {"reference", "modifiedDate"}
def test_merge_allof_annotation_keywords_do_not_block_merge():
# Codat's PagingInfo carries `definitions` alongside its properties. Treating
# any unrecognized keyword as disqualifying left those compositions unmerged,
# which is what kept the response model empty.
schema = {
"title": "Paged results",
"x-internal": True,
"allOf": [
{"type": "object", "properties": {"results": {"type": "array"}}},
{
"title": "Pagination information",
"properties": {"pageNumber": {"type": "integer"}},
"definitions": {"links": {"type": "object"}},
"additionalProperties": False,
},
],
}
merged = merge_allof(schema)
assert set(merged["properties"]) == {"results", "pageNumber"}
assert merged["type"] == "object"
def test_merge_allof_blocks_on_structural_keywords():
# enum, const and items are scalar/array constraints, not property bags.
for blocking in ({"enum": ["a"]}, {"const": 1}, {"items": {"type": "string"}}):
schema = {"allOf": [{"properties": {"a": {}}}, blocking]}
assert merge_allof(schema) == schema
def test_merge_allof_leaves_instance_data_untouched():
"""``default`` / ``example`` / ``enum`` hold instance data, not subschemas.
A default value that happens to contain an ``allOf`` key was rewritten into
``{"type": "object"}``, destroying the value.
"""
schema = {
"type": "object",
"properties": {
"cfg": {
"type": "object",
"default": {"allOf": [{"a": 1}, {"b": 2}]},
"example": {"allOf": [{"p": 1}]},
},
"mode": {"type": "string", "enum": [{"allOf": [{"q": 1}]}]},
},
"x-vendor": {"allOf": [{"z": 1}, {"w": 2}]},
}
merged = merge_allof(schema)
assert merged["properties"]["cfg"]["default"] == {"allOf": [{"a": 1}, {"b": 2}]}
assert merged["properties"]["cfg"]["example"] == {"allOf": [{"p": 1}]}
assert merged["properties"]["mode"]["enum"] == [{"allOf": [{"q": 1}]}]
assert merged["x-vendor"] == {"allOf": [{"z": 1}, {"w": 2}]}
def test_deref_schema_leaves_instance_data_untouched():
"""A ``$ref`` key inside a default value is a literal, not a reference."""
spec = {"components": {"schemas": {"Thing": {"type": "integer"}}}}
node = {
"type": "object",
"properties": {
"payload": {
"type": "object",
"default": {"$ref": "#/components/schemas/Thing"},
},
"real": {"$ref": "#/components/schemas/Thing"},
},
"x-sample": {"$ref": "#/components/schemas/Thing"},
}
out = deref_schema(spec, node)
# The literal survives, the genuine reference resolves.
assert out["properties"]["payload"]["default"] == {
"$ref": "#/components/schemas/Thing"
}
assert out["x-sample"] == {"$ref": "#/components/schemas/Thing"}
assert out["properties"]["real"] == {"type": "integer"}
def _serve_documents(monkeypatch, documents):
from openbb_cli.dispatchers import openapi_schema

184
cli/tests/test_dispatchers_spec.py
View File

@@ -1342,3 +1342,187 @@ def test_add_normalized_parameter_with_datetime_param_uses_coercer():
)
action = next(a for a in p._actions if a.dest == "fromDateTime")
assert action.type is _coerce_iso_datetime
def test_build_command_spec_inherits_path_item_parameters():
"""Path and query parameters declared on the path item apply to the operation."""
openapi = {
"paths": {
"/api/v1/x/{record_id}": {
"parameters": [
{"name": "record_id", "in": "path", "required": True},
{"name": "page", "in": "query", "schema": {"type": "integer"}},
],
"get": {
"operationId": "x",
"parameters": [{"name": "symbol", "schema": {"type": "string"}}],
},
}
}
}
out = build_command_spec(openapi)
names = [p["name"] for p in out["x"]["parameters"]]
assert names == ["record_id", "page", "symbol"]
record = next(p for p in out["x"]["parameters"] if p["name"] == "record_id")
assert record["required"] is True
def test_build_command_spec_routes_required_path_item_headers_to_credentials():
"""A required path-item header/cookie is kept as a credential, not a CLI arg.
An optional one (``session`` here) still has nowhere to go and is dropped,
same as before — only a required value is the "every operation needs
this static secret" profile that credential handling exists for.
"""
openapi = {
"paths": {
"/api/v1/x": {
"parameters": [
{"name": "x-tenant-id", "in": "header", "required": True},
{"name": "x-session", "in": "cookie", "required": True},
{"name": "session", "in": "cookie"},
{"name": "page", "in": "query", "schema": {"type": "integer"}},
],
"get": {"operationId": "x"},
}
}
}
out = build_command_spec(openapi)
params = {p["name"]: p for p in out["x"]["parameters"]}
assert set(params) == {"page", "x-tenant-id", "x-session"}
assert params["x-tenant-id"]["in"] == "header"
assert params["x-tenant-id"]["required"] is True
assert params["x-session"]["in"] == "cookie"
assert params["x-session"]["required"] is True
def test_build_command_spec_resolves_path_item_ref_parameters():
"""A path-item parameter given as a $ref is resolved before inheriting."""
openapi = {
"components": {
"parameters": {
"companyId": {
"name": "companyId",
"in": "path",
"required": True,
"schema": {"type": "string"},
}
}
},
"paths": {
"/api/v1/accounts/{companyId}": {
"parameters": [{"$ref": "#/components/parameters/companyId"}],
"get": {"operationId": "accounts"},
}
},
}
out = build_command_spec(openapi)
assert [p["name"] for p in out["accounts"]["parameters"]] == ["companyId"]
def test_build_command_spec_operation_parameter_overrides_path_item():
"""An operation parameter wins over an inherited one with the same name and location."""
openapi = {
"paths": {
"/api/v1/x": {
"parameters": [
{"name": "limit", "in": "query", "schema": {"type": "string"}}
],
"get": {
"operationId": "x",
"parameters": [
{
"name": "limit",
"in": "query",
"required": True,
"schema": {"type": "integer"},
}
],
},
}
}
}
out = build_command_spec(openapi)
params = out["x"]["parameters"]
assert len(params) == 1
assert params[0]["type"] == "integer"
assert params[0]["required"] is True
def test_build_command_spec_path_item_param_kept_when_location_differs():
"""Overriding is keyed on (name, in) — a same-named param elsewhere is kept."""
openapi = {
"paths": {
"/api/v1/x": {
"parameters": [
{"name": "id", "in": "path", "schema": {"type": "string"}}
],
"get": {
"operationId": "x",
"parameters": [
{"name": "id", "in": "query", "schema": {"type": "string"}}
],
},
}
}
}
out = build_command_spec(openapi)
locations = sorted(p["in"] for p in out["x"]["parameters"] if p["name"] == "id")
assert locations == ["path", "query"]
def test_build_command_spec_path_item_parameters_apply_to_every_operation():
"""Both operations under one path item inherit its shared parameters."""
openapi = {
"paths": {
"/api/v1/x": {
"parameters": [
{"name": "page", "in": "query", "schema": {"type": "integer"}}
],
"get": {"operationId": "x"},
},
"/api/v1/y": {
"parameters": [
{"name": "page", "in": "query", "schema": {"type": "integer"}}
],
"post": {"operationId": "y"},
},
}
}
out = build_command_spec(openapi)
assert [p["name"] for p in out["x"]["parameters"]] == ["page"]
assert [p["name"] for p in out["y"]["parameters"]] == ["page"]
def test_build_command_spec_skips_unresolvable_path_item_ref_param():
"""An unresolvable path-item $ref is dropped rather than crashing."""
openapi = {
"paths": {
"/api/v1/x": {
"parameters": [{"$ref": "#/components/parameters/missing"}],
"get": {
"operationId": "x",
"parameters": [{"name": "kept", "schema": {"type": "string"}}],
},
}
}
}
out = build_command_spec(openapi)
assert [p["name"] for p in out["x"]["parameters"]] == ["kept"]
def test_build_command_spec_ignores_malformed_path_item_parameters():
"""Non-dict path-item parameter entries are skipped defensively."""
openapi = {
"paths": {
"/api/v1/x": {
"parameters": ["not-a-dict", {"in": "query"}],
"get": {
"operationId": "x",
"parameters": [{"name": "kept", "schema": {"type": "string"}}],
},
}
}
}
out = build_command_spec(openapi)
assert [p["name"] for p in out["x"]["parameters"]] == ["kept"]