From 98b4f4fb0723202a516e0dd4924b6ae45c92a9e7 Mon Sep 17 00:00:00 2001 From: Iddo Date: Sun, 10 May 2026 07:17:57 +0200 Subject: [PATCH 1/7] fix(ctl): handle extensions paths in display_schema_load_errors (#1007) --- infrahub_sdk/ctl/schema.py | 117 ++++++++++++++++++++++------------ tests/unit/sdk/test_schema.py | 93 +++++++++++++++++++++++++++ 2 files changed, 171 insertions(+), 39 deletions(-) diff --git a/infrahub_sdk/ctl/schema.py b/infrahub_sdk/ctl/schema.py index 67bd9923..1769149f 100644 --- a/infrahub_sdk/ctl/schema.py +++ b/infrahub_sdk/ctl/schema.py @@ -59,44 +59,65 @@ def display_schema_load_errors(response: dict[str, Any], schemas_data: list[Sche loc_path = error.get("loc", []) if not valid_error_path(loc_path=loc_path): continue - - # if the len of the path is equal to 6, the error is at the root of the object - # if the len of the path is higher than 6, the error is in an attribute or a relationships - schema_index = int(loc_path[2]) + _render_schema_error(error=error, loc_path=loc_path, schemas_data=schemas_data) + + +def _render_schema_error(error: dict[str, Any], loc_path: list[Any], schemas_data: list[SchemaFile]) -> None: + # Two layout shapes for loc_path. tail is the part after the node index. + # Top-level: body / schemas / / (nodes|generics) / / [ / ] + # Extensions: body / schemas / / extensions / (nodes|generics|relationships) / / [ / ] + schema_index = int(loc_path[2]) + is_extension = loc_path[3] == "extensions" + if is_extension: + container = loc_path[4] + node_index = int(loc_path[5]) + tail = loc_path[6:] + else: + container = loc_path[3] node_index = int(loc_path[4]) - node = get_node(schemas_data=schemas_data, schema_index=schema_index, node_index=node_index) + tail = loc_path[5:] + + node = get_node( + schemas_data=schemas_data, + schema_index=schema_index, + node_index=node_index, + container=container, + is_extension=is_extension, + ) - if not node: - console.print("Node data not found.") - continue + if not node: + console.print("Node data not found.") + return - if len(loc_path) == 6: - loc_type = loc_path[-1] - input_str = error.get("input", None) - error_message = f"{loc_type} ({input_str}) | {error['msg']} ({error['type']})" - console.print( - f" Node: {node.get('namespace', None)}{node.get('name', None)} | {error_message}", markup=False - ) + node_label = ( + (node.get("kind") or node.get("name") or "") + if is_extension + else f"{node.get('namespace', None)}{node.get('name', None)}" + ) + path_suffix = f" (extensions/{container})" if is_extension else "" + input_str = error.get("input") + + if len(tail) == 1: + loc_type = tail[0] + error_message = f"{loc_type} ({input_str}) | {error['msg']} ({error['type']})" + elif len(tail) > 1: + loc_type = tail[0] + attribute = tail[1] + input_label = _resolve_attribute_label(error_data=node.get(loc_type, []), attribute=attribute) + error_message = f"{loc_type[:-1].title()}: {input_label} ({input_str}) | {error['msg']} ({error['type']})" + else: + return - elif len(loc_path) > 6: - loc_type = loc_path[5] - error_data = node[loc_type] - attribute = loc_path[6] - - if isinstance(attribute, str): - input_label = None - for data in error_data: - if data.get(attribute) is not None: - input_label = data.get("name", None) - break - else: - input_label = error_data[attribute].get("name", None) + console.print(f" Node: {node_label}{path_suffix} | {error_message}", markup=False) - input_str = error.get("input", None) - error_message = f"{loc_type[:-1].title()}: {input_label} ({input_str}) | {error['msg']} ({error['type']})" - console.print( - f" Node: {node.get('namespace', None)}{node.get('name', None)} | {error_message}", markup=False - ) + +def _resolve_attribute_label(error_data: list[dict[str, Any]], attribute: Any) -> str | None: + if isinstance(attribute, str): + for data in error_data: + if data.get(attribute) is not None: + return data.get("name", None) + return None + return error_data[attribute].get("name", None) def handle_non_detail_errors(response: dict[str, Any]) -> None: @@ -110,12 +131,30 @@ def handle_non_detail_errors(response: dict[str, Any]) -> None: def valid_error_path(loc_path: list[Any]) -> bool: - return len(loc_path) >= 6 and loc_path[0] == "body" and loc_path[1] == "schemas" - - -def get_node(schemas_data: list[SchemaFile], schema_index: int, node_index: int) -> dict | None: - if schema_index < len(schemas_data) and node_index < len(schemas_data[schema_index].payload["nodes"]): - return schemas_data[schema_index].payload["nodes"][node_index] + if len(loc_path) < 6 or loc_path[0] != "body" or loc_path[1] != "schemas" or not isinstance(loc_path[2], int): + return False + if loc_path[3] == "extensions": + return ( + len(loc_path) >= 7 + and loc_path[4] in {"nodes", "generics", "relationships"} + and isinstance(loc_path[5], int) + ) + return loc_path[3] in {"nodes", "generics"} and isinstance(loc_path[4], int) + + +def get_node( + schemas_data: list[SchemaFile], + schema_index: int, + node_index: int, + container: str = "nodes", + is_extension: bool = False, +) -> dict | None: + if schema_index >= len(schemas_data): + return None + payload = schemas_data[schema_index].payload + items = payload.get("extensions", {}).get(container, []) if is_extension else payload.get(container, []) + if node_index < len(items): + return items[node_index] return None diff --git a/tests/unit/sdk/test_schema.py b/tests/unit/sdk/test_schema.py index 308fe64a..724e5dcd 100644 --- a/tests/unit/sdk/test_schema.py +++ b/tests/unit/sdk/test_schema.py @@ -490,6 +490,99 @@ async def test_display_schema_load_errors_details_when_error_is_in_attribute_or_ assert output == expected_console +@mock.patch( + "infrahub_sdk.ctl.schema.get_node", + return_value={"kind": "DcimGenericDevice", "include_in_menu": False}, +) +async def test_display_schema_load_errors_details_extensions_top_level(mock_get_node: MagicMock) -> None: + """Validate error rendering when the failing path is inside an extensions block at field level.""" + error = { + "detail": [ + { + "type": "extra_forbidden", + "loc": ["body", "schemas", 0, "extensions", "generics", 0, "include_in_menu"], + "msg": "Extra inputs are not permitted", + "input": False, + }, + ] + } + + with mock.patch("infrahub_sdk.ctl.schema.console", Console(file=StringIO(), width=1000)) as console: + display_schema_load_errors(response=error, schemas_data=[]) + mock_get_node.assert_called_once() + output = console.file.getvalue() + expected_console = """Unable to load the schema: + Node: DcimGenericDevice (extensions/generics) | include_in_menu (False) | Extra inputs are not permitted (extra_forbidden) +""" + assert output == expected_console + + +@mock.patch( + "infrahub_sdk.ctl.schema.get_node", + return_value={ + "kind": "DcimGenericDevice", + "attributes": [{"name": "speed", "kind": "Number", "min_value": 0}], + }, +) +async def test_display_schema_load_errors_details_extensions_nested_attribute(mock_get_node: MagicMock) -> None: + """Validate error rendering for a nested attribute error inside an extensions block.""" + error = { + "detail": [ + { + "type": "extra_forbidden", + "loc": ["body", "schemas", 0, "extensions", "generics", 0, "attributes", "min_value"], + "msg": "Extra inputs are not permitted", + "input": 0, + }, + ] + } + + with mock.patch("infrahub_sdk.ctl.schema.console", Console(file=StringIO(), width=1000)) as console: + display_schema_load_errors(response=error, schemas_data=[]) + mock_get_node.assert_called_once() + output = console.file.getvalue() + expected_console = """Unable to load the schema: + Node: DcimGenericDevice (extensions/generics) | Attribute: speed (0) | Extra inputs are not permitted (extra_forbidden) +""" + assert output == expected_console + + +async def test_display_schema_load_errors_skips_unknown_path_shapes() -> None: + """Unknown or malformed loc_path shapes are skipped silently, never crash the renderer.""" + error = { + "detail": [ + # Wrong root prefix + {"type": "value_error", "loc": ["body", "headers", "x-test"], "msg": "bad", "input": "x"}, + # schema_index is not an int + { + "type": "value_error", + "loc": ["body", "schemas", "not-an-int", "nodes", 0, "name"], + "msg": "bad", + "input": "x", + }, + # Unknown container + { + "type": "value_error", + "loc": ["body", "schemas", 0, "wat", 0, "name"], + "msg": "bad", + "input": "x", + }, + # Extensions path with non-int node index — was the original crash + { + "type": "value_error", + "loc": ["body", "schemas", 0, "extensions", "generics", "include_in_menu"], + "msg": "bad", + "input": "x", + }, + ] + } + + with mock.patch("infrahub_sdk.ctl.schema.console", Console(file=StringIO(), width=1000)) as console: + display_schema_load_errors(response=error, schemas_data=[]) + output = console.file.getvalue() + assert output == "Unable to load the schema:\n" + + def test_schema_base__get_schema_name__returns_correct_schema_name_for_protocols() -> None: assert InfrahubSchemaBase._get_schema_name(schema=BuiltinTagSync) == "BuiltinTag" assert InfrahubSchemaBase._get_schema_name(schema=BuiltinTag) == "BuiltinTag" From 0ab1e4d4cacc04efcb78c3d3b196747a985223b8 Mon Sep 17 00:00:00 2001 From: Iddo Date: Sun, 10 May 2026 11:10:51 +0200 Subject: [PATCH 2/7] chore(changelog): add newsfragment for #1007 Co-Authored-By: Claude Opus 4.7 (1M context) --- changelog/1007.fixed.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/1007.fixed.md diff --git a/changelog/1007.fixed.md b/changelog/1007.fixed.md new file mode 100644 index 00000000..d87d256a --- /dev/null +++ b/changelog/1007.fixed.md @@ -0,0 +1 @@ +Render schema rejections originating in an `extensions:` block as a readable one-line message in `infrahubctl schema load`, instead of crashing with `ValueError: invalid literal for int()`. From ea5d6d020328b86060df242821aeeb5994b103df Mon Sep 17 00:00:00 2001 From: Iddo Date: Tue, 12 May 2026 07:25:24 +0200 Subject: [PATCH 3/7] test(ctl): replace brittle schema-load-error mocks with integration tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace patch()-heavy unit tests for display_schema_load_errors with integration tests that exercise the real server response → renderer path, so we catch any drift between the schema-load endpoint's error shape and CLI handling (particularly for extensions paths). Add an optional Console parameter to display_schema_load_errors so tests can capture rendered output via dependency injection instead of patching the module-level console. Keep valid_error_path covered as a fast parametrized unit test with no patchin --- infrahub_sdk/ctl/schema.py | 28 +++++--- tests/integration/test_schema.py | 73 +++++++++++++++++++ tests/unit/sdk/test_schema.py | 116 +++++++------------------------ 3 files changed, 116 insertions(+), 101 deletions(-) diff --git a/infrahub_sdk/ctl/schema.py b/infrahub_sdk/ctl/schema.py index 1769149f..2aac4bc6 100644 --- a/infrahub_sdk/ctl/schema.py +++ b/infrahub_sdk/ctl/schema.py @@ -49,20 +49,25 @@ def validate_schema_content_and_exit(client: InfrahubClient, schemas: list[Schem raise typer.Exit(1) -def display_schema_load_errors(response: dict[str, Any], schemas_data: list[SchemaFile]) -> None: - console.print("[red]Unable to load the schema:") +def display_schema_load_errors( + response: dict[str, Any], schemas_data: list[SchemaFile], output: Console | None = None +) -> None: + out = output or console + out.print("[red]Unable to load the schema:") if "detail" not in response: - handle_non_detail_errors(response=response) + handle_non_detail_errors(response=response, output=out) return for error in response["detail"]: loc_path = error.get("loc", []) if not valid_error_path(loc_path=loc_path): continue - _render_schema_error(error=error, loc_path=loc_path, schemas_data=schemas_data) + _render_schema_error(error=error, loc_path=loc_path, schemas_data=schemas_data, output=out) -def _render_schema_error(error: dict[str, Any], loc_path: list[Any], schemas_data: list[SchemaFile]) -> None: +def _render_schema_error( + error: dict[str, Any], loc_path: list[Any], schemas_data: list[SchemaFile], output: Console +) -> None: # Two layout shapes for loc_path. tail is the part after the node index. # Top-level: body / schemas / / (nodes|generics) / / [ / ] # Extensions: body / schemas / / extensions / (nodes|generics|relationships) / / [ / ] @@ -86,7 +91,7 @@ def _render_schema_error(error: dict[str, Any], loc_path: list[Any], schemas_dat ) if not node: - console.print("Node data not found.") + output.print("Node data not found.") return node_label = ( @@ -108,7 +113,7 @@ def _render_schema_error(error: dict[str, Any], loc_path: list[Any], schemas_dat else: return - console.print(f" Node: {node_label}{path_suffix} | {error_message}", markup=False) + output.print(f" Node: {node_label}{path_suffix} | {error_message}", markup=False) def _resolve_attribute_label(error_data: list[dict[str, Any]], attribute: Any) -> str | None: @@ -120,14 +125,15 @@ def _resolve_attribute_label(error_data: list[dict[str, Any]], attribute: Any) - return error_data[attribute].get("name", None) -def handle_non_detail_errors(response: dict[str, Any]) -> None: +def handle_non_detail_errors(response: dict[str, Any], output: Console | None = None) -> None: + out = output or console if "error" in response: - console.print(f" {response.get('error')}") + out.print(f" {response.get('error')}") elif "errors" in response: for error in response["errors"]: - console.print(f" {error.get('message')}") + out.print(f" {error.get('message')}") else: - console.print(f" '{response}'") + out.print(f" '{response}'") def valid_error_path(loc_path: list[Any]) -> bool: diff --git a/tests/integration/test_schema.py b/tests/integration/test_schema.py index 314a3ffa..a0eee32b 100644 --- a/tests/integration/test_schema.py +++ b/tests/integration/test_schema.py @@ -1,11 +1,16 @@ +from io import StringIO +from pathlib import Path from typing import Any import pytest +from rich.console import Console from infrahub_sdk import InfrahubClient +from infrahub_sdk.ctl.schema import display_schema_load_errors from infrahub_sdk.exceptions import BranchNotFoundError from infrahub_sdk.schema import NodeSchemaAPI from infrahub_sdk.testing.docker import TestInfrahubDockerClient +from infrahub_sdk.yaml import SchemaFile class TestInfrahubSchema(TestInfrahubDockerClient): @@ -43,3 +48,71 @@ async def test_schema_load_many( schema_nodes = await client.schema.all(refresh=True) assert "InfraRack" in schema_nodes assert "ProcurementContract" in schema_nodes + + +class TestInfrahubSchemaLoadErrorRendering(TestInfrahubDockerClient): + """Render real server error responses through display_schema_load_errors. + + These exist as integration tests so we catch any drift between the server's + validation error payload shape and the CLI renderer, particularly for + `extensions` paths which previously went unhandled. + """ + + async def test_extension_top_level_field_error(self, client: InfrahubClient) -> None: + broken_schema = { + "version": "1.0", + "extensions": { + "nodes": [ + { + "kind": "BuiltinTag", + "namespace": "Forbidden", + } + ] + }, + } + + response = await client.schema.load(schemas=[broken_schema]) + + assert response.errors, "Server should reject a forbidden field on an extensions/nodes entry" + assert "detail" in response.errors + + schemas_data = [SchemaFile(location=Path("broken.yml"), content=broken_schema)] + buffer = StringIO() + output = Console(file=buffer, width=1000, force_terminal=False) + display_schema_load_errors(response=response.errors, schemas_data=schemas_data, output=output) + rendered = buffer.getvalue() + + assert "Unable to load the schema" in rendered + assert "BuiltinTag" in rendered + assert "extensions/nodes" in rendered + + async def test_extension_nested_attribute_error(self, client: InfrahubClient) -> None: + broken_schema = { + "version": "1.0", + "extensions": { + "nodes": [ + { + "kind": "BuiltinTag", + "attributes": [ + {"name": "speed", "kind": "Number", "made_up": True}, + ], + } + ] + }, + } + + response = await client.schema.load(schemas=[broken_schema]) + + assert response.errors, "Server should reject a forbidden field on an extensions attribute entry" + assert "detail" in response.errors + + schemas_data = [SchemaFile(location=Path("broken.yml"), content=broken_schema)] + buffer = StringIO() + output = Console(file=buffer, width=1000, force_terminal=False) + display_schema_load_errors(response=response.errors, schemas_data=schemas_data, output=output) + rendered = buffer.getvalue() + + assert "Unable to load the schema" in rendered + assert "BuiltinTag" in rendered + assert "extensions/nodes" in rendered + assert "speed" in rendered diff --git a/tests/unit/sdk/test_schema.py b/tests/unit/sdk/test_schema.py index 724e5dcd..fe390537 100644 --- a/tests/unit/sdk/test_schema.py +++ b/tests/unit/sdk/test_schema.py @@ -8,7 +8,7 @@ from rich.console import Console from infrahub_sdk import Config, InfrahubClient, InfrahubClientSync -from infrahub_sdk.ctl.schema import display_schema_load_errors +from infrahub_sdk.ctl.schema import display_schema_load_errors, valid_error_path from infrahub_sdk.exceptions import SchemaNotFoundError, ValidationError from infrahub_sdk.protocols import BuiltinIPAddress, BuiltinIPAddressSync, BuiltinTag, BuiltinTagSync from infrahub_sdk.schema import BranchSchema, InfrahubSchema, InfrahubSchemaBase, InfrahubSchemaSync, NodeSchemaAPI @@ -490,97 +490,33 @@ async def test_display_schema_load_errors_details_when_error_is_in_attribute_or_ assert output == expected_console -@mock.patch( - "infrahub_sdk.ctl.schema.get_node", - return_value={"kind": "DcimGenericDevice", "include_in_menu": False}, +@pytest.mark.parametrize( + "loc_path", + [ + pytest.param(["body", "schemas", 0, "nodes", 0, "name"], id="top-level-nodes"), + pytest.param(["body", "schemas", 0, "generics", 1, "attributes", 0], id="top-level-generics"), + pytest.param(["body", "schemas", 0, "extensions", "nodes", 0, "kind"], id="extension-nodes"), + pytest.param(["body", "schemas", 0, "extensions", "generics", 0, "name"], id="extension-generics"), + pytest.param(["body", "schemas", 0, "extensions", "relationships", 2, "peer"], id="extension-relationships"), + ], ) -async def test_display_schema_load_errors_details_extensions_top_level(mock_get_node: MagicMock) -> None: - """Validate error rendering when the failing path is inside an extensions block at field level.""" - error = { - "detail": [ - { - "type": "extra_forbidden", - "loc": ["body", "schemas", 0, "extensions", "generics", 0, "include_in_menu"], - "msg": "Extra inputs are not permitted", - "input": False, - }, - ] - } - - with mock.patch("infrahub_sdk.ctl.schema.console", Console(file=StringIO(), width=1000)) as console: - display_schema_load_errors(response=error, schemas_data=[]) - mock_get_node.assert_called_once() - output = console.file.getvalue() - expected_console = """Unable to load the schema: - Node: DcimGenericDevice (extensions/generics) | include_in_menu (False) | Extra inputs are not permitted (extra_forbidden) -""" - assert output == expected_console - - -@mock.patch( - "infrahub_sdk.ctl.schema.get_node", - return_value={ - "kind": "DcimGenericDevice", - "attributes": [{"name": "speed", "kind": "Number", "min_value": 0}], - }, +def test_valid_error_path_accepts_known_shapes(loc_path: list) -> None: + assert valid_error_path(loc_path=loc_path) + + +@pytest.mark.parametrize( + "loc_path", + [ + pytest.param(["body", "headers", "x-test"], id="wrong-root"), + pytest.param(["body", "schemas", "not-an-int", "nodes", 0, "name"], id="non-int-schema-index"), + pytest.param(["body", "schemas", 0, "wat", 0, "name"], id="unknown-container"), + pytest.param(["body", "schemas", 0, "extensions", "generics", "include_in_menu"], id="non-int-extension-index"), + pytest.param(["body", "schemas", 0, "extensions", "wat", 0, "name"], id="unknown-extension-container"), + pytest.param(["body", "schemas", 0, "nodes"], id="too-short"), + ], ) -async def test_display_schema_load_errors_details_extensions_nested_attribute(mock_get_node: MagicMock) -> None: - """Validate error rendering for a nested attribute error inside an extensions block.""" - error = { - "detail": [ - { - "type": "extra_forbidden", - "loc": ["body", "schemas", 0, "extensions", "generics", 0, "attributes", "min_value"], - "msg": "Extra inputs are not permitted", - "input": 0, - }, - ] - } - - with mock.patch("infrahub_sdk.ctl.schema.console", Console(file=StringIO(), width=1000)) as console: - display_schema_load_errors(response=error, schemas_data=[]) - mock_get_node.assert_called_once() - output = console.file.getvalue() - expected_console = """Unable to load the schema: - Node: DcimGenericDevice (extensions/generics) | Attribute: speed (0) | Extra inputs are not permitted (extra_forbidden) -""" - assert output == expected_console - - -async def test_display_schema_load_errors_skips_unknown_path_shapes() -> None: - """Unknown or malformed loc_path shapes are skipped silently, never crash the renderer.""" - error = { - "detail": [ - # Wrong root prefix - {"type": "value_error", "loc": ["body", "headers", "x-test"], "msg": "bad", "input": "x"}, - # schema_index is not an int - { - "type": "value_error", - "loc": ["body", "schemas", "not-an-int", "nodes", 0, "name"], - "msg": "bad", - "input": "x", - }, - # Unknown container - { - "type": "value_error", - "loc": ["body", "schemas", 0, "wat", 0, "name"], - "msg": "bad", - "input": "x", - }, - # Extensions path with non-int node index — was the original crash - { - "type": "value_error", - "loc": ["body", "schemas", 0, "extensions", "generics", "include_in_menu"], - "msg": "bad", - "input": "x", - }, - ] - } - - with mock.patch("infrahub_sdk.ctl.schema.console", Console(file=StringIO(), width=1000)) as console: - display_schema_load_errors(response=error, schemas_data=[]) - output = console.file.getvalue() - assert output == "Unable to load the schema:\n" +def test_valid_error_path_rejects_unknown_shapes(loc_path: list) -> None: + assert not valid_error_path(loc_path=loc_path) def test_schema_base__get_schema_name__returns_correct_schema_name_for_protocols() -> None: From b78e5468dd283bf0baa44fa43e2323840fed3a51 Mon Sep 17 00:00:00 2001 From: Iddo Date: Tue, 12 May 2026 07:27:30 +0200 Subject: [PATCH 4/7] refactor(ctl): type schema get_node container as Literal Replace on with a Literal of nodes, generics, relationships so the accepted values are explicit at the type level. Matches the set already enforced at runtime by valid_error_path --- infrahub_sdk/ctl/schema.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/infrahub_sdk/ctl/schema.py b/infrahub_sdk/ctl/schema.py index 2aac4bc6..7ead1014 100644 --- a/infrahub_sdk/ctl/schema.py +++ b/infrahub_sdk/ctl/schema.py @@ -4,7 +4,7 @@ import time from datetime import datetime, timezone from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Literal import typer import yaml @@ -148,11 +148,14 @@ def valid_error_path(loc_path: list[Any]) -> bool: return loc_path[3] in {"nodes", "generics"} and isinstance(loc_path[4], int) +SchemaContainer = Literal["nodes", "generics", "relationships"] + + def get_node( schemas_data: list[SchemaFile], schema_index: int, node_index: int, - container: str = "nodes", + container: SchemaContainer = "nodes", is_extension: bool = False, ) -> dict | None: if schema_index >= len(schemas_data): From 1f85ac51129d7f6fd928a772dd19b146dc24e2e2 Mon Sep 17 00:00:00 2001 From: Iddo Date: Tue, 12 May 2026 07:29:04 +0200 Subject: [PATCH 5/7] fix(ctl): default missing msg/type keys in schema error renderer --- infrahub_sdk/ctl/schema.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/infrahub_sdk/ctl/schema.py b/infrahub_sdk/ctl/schema.py index 7ead1014..6dae28f1 100644 --- a/infrahub_sdk/ctl/schema.py +++ b/infrahub_sdk/ctl/schema.py @@ -101,15 +101,17 @@ def _render_schema_error( ) path_suffix = f" (extensions/{container})" if is_extension else "" input_str = error.get("input") + err_msg = error.get("msg", "No error message") + err_type = error.get("type", "unknown") if len(tail) == 1: loc_type = tail[0] - error_message = f"{loc_type} ({input_str}) | {error['msg']} ({error['type']})" + error_message = f"{loc_type} ({input_str}) | {err_msg} ({err_type})" elif len(tail) > 1: loc_type = tail[0] attribute = tail[1] input_label = _resolve_attribute_label(error_data=node.get(loc_type, []), attribute=attribute) - error_message = f"{loc_type[:-1].title()}: {input_label} ({input_str}) | {error['msg']} ({error['type']})" + error_message = f"{loc_type[:-1].title()}: {input_label} ({input_str}) | {err_msg} ({err_type})" else: return From 72947dc77ecabe20dff57c997ffa96cffa2de294 Mon Sep 17 00:00:00 2001 From: Iddo Date: Tue, 12 May 2026 07:38:34 +0200 Subject: [PATCH 6/7] fix(ctl): guard out-of-range index in _resolve_attribute_label and annotate _render_schema_error parsing branches --- infrahub_sdk/ctl/schema.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/infrahub_sdk/ctl/schema.py b/infrahub_sdk/ctl/schema.py index 6dae28f1..13b5d3e6 100644 --- a/infrahub_sdk/ctl/schema.py +++ b/infrahub_sdk/ctl/schema.py @@ -94,6 +94,7 @@ def _render_schema_error( output.print("Node data not found.") return + # Extensions reference an existing node by `kind`; new top-level nodes are identified by `namespace+name`. node_label = ( (node.get("kind") or node.get("name") or "") if is_extension @@ -105,12 +106,16 @@ def _render_schema_error( err_type = error.get("type", "unknown") if len(tail) == 1: + # Error on a direct field of the node (e.g. `name`, `namespace`). loc_type = tail[0] error_message = f"{loc_type} ({input_str}) | {err_msg} ({err_type})" elif len(tail) > 1: + # Error nested inside a collection (e.g. attributes[2].kind, relationships[0].peer). + # loc_type is the collection name; attribute is either its index or the failing field name. loc_type = tail[0] attribute = tail[1] input_label = _resolve_attribute_label(error_data=node.get(loc_type, []), attribute=attribute) + # Trim the trailing 's' so "attributes" → "Attribute" in the rendered label. error_message = f"{loc_type[:-1].title()}: {input_label} ({input_str}) | {err_msg} ({err_type})" else: return @@ -124,7 +129,9 @@ def _resolve_attribute_label(error_data: list[dict[str, Any]], attribute: Any) - if data.get(attribute) is not None: return data.get("name", None) return None - return error_data[attribute].get("name", None) + if isinstance(attribute, int) and 0 <= attribute < len(error_data): + return error_data[attribute].get("name", None) + return None def handle_non_detail_errors(response: dict[str, Any], output: Console | None = None) -> None: From 3a0584a1df499fc82d88f2246810f0cd0ec8b7a0 Mon Sep 17 00:00:00 2001 From: Iddo Date: Tue, 12 May 2026 21:13:18 +0200 Subject: [PATCH 7/7] refactor(ctl): address schema-load review feedback and use rich console.capture() and hoist SchemaContainer alias --- infrahub_sdk/ctl/schema.py | 5 ++--- tests/integration/test_schema.py | 17 ++++++++--------- 2 files changed, 10 insertions(+), 12 deletions(-) diff --git a/infrahub_sdk/ctl/schema.py b/infrahub_sdk/ctl/schema.py index 13b5d3e6..21f9c7a9 100644 --- a/infrahub_sdk/ctl/schema.py +++ b/infrahub_sdk/ctl/schema.py @@ -24,6 +24,8 @@ if TYPE_CHECKING: from .. import InfrahubClient +SchemaContainer = Literal["nodes", "generics", "relationships"] + app = AsyncTyper() console = Console() @@ -157,9 +159,6 @@ def valid_error_path(loc_path: list[Any]) -> bool: return loc_path[3] in {"nodes", "generics"} and isinstance(loc_path[4], int) -SchemaContainer = Literal["nodes", "generics", "relationships"] - - def get_node( schemas_data: list[SchemaFile], schema_index: int, diff --git a/tests/integration/test_schema.py b/tests/integration/test_schema.py index a0eee32b..42307e3f 100644 --- a/tests/integration/test_schema.py +++ b/tests/integration/test_schema.py @@ -1,4 +1,3 @@ -from io import StringIO from pathlib import Path from typing import Any @@ -77,10 +76,10 @@ async def test_extension_top_level_field_error(self, client: InfrahubClient) -> assert "detail" in response.errors schemas_data = [SchemaFile(location=Path("broken.yml"), content=broken_schema)] - buffer = StringIO() - output = Console(file=buffer, width=1000, force_terminal=False) - display_schema_load_errors(response=response.errors, schemas_data=schemas_data, output=output) - rendered = buffer.getvalue() + console = Console(width=1000) + with console.capture() as capture: + display_schema_load_errors(response=response.errors, schemas_data=schemas_data, output=console) + rendered = capture.get() assert "Unable to load the schema" in rendered assert "BuiltinTag" in rendered @@ -107,10 +106,10 @@ async def test_extension_nested_attribute_error(self, client: InfrahubClient) -> assert "detail" in response.errors schemas_data = [SchemaFile(location=Path("broken.yml"), content=broken_schema)] - buffer = StringIO() - output = Console(file=buffer, width=1000, force_terminal=False) - display_schema_load_errors(response=response.errors, schemas_data=schemas_data, output=output) - rendered = buffer.getvalue() + console = Console(width=1000) + with console.capture() as capture: + display_schema_load_errors(response=response.errors, schemas_data=schemas_data, output=console) + rendered = capture.get() assert "Unable to load the schema" in rendered assert "BuiltinTag" in rendered