From ee56603155fca67be123b03828018dae3fdffeb9 Mon Sep 17 00:00:00 2001 From: Bearchitek Date: Wed, 18 Feb 2026 15:06:22 +0100 Subject: [PATCH 1/6] Add infrahubctl schema export command Adds `infrahubctl schema export` to fetch user-defined schemas from an Infrahub server and write them as YAML files (one per namespace). Restricted namespaces (Core, Builtin, Internal, etc.) and auto-generated types (ProfileSchemaAPI, TemplateSchemaAPI) are excluded by default. RESTRICTED_NAMESPACES is mirrored from the Infrahub server constants so the server can later consume it from the SDK. Co-Authored-By: Claude Sonnet 4.6 --- infrahub_sdk/constants.py | 18 ++ infrahub_sdk/ctl/schema.py | 90 ++++++++- tests/unit/ctl/test_schema_export.py | 288 +++++++++++++++++++++++++++ 3 files changed, 395 insertions(+), 1 deletion(-) create mode 100644 tests/unit/ctl/test_schema_export.py diff --git a/infrahub_sdk/constants.py b/infrahub_sdk/constants.py index 04dd6b95..b3a53783 100644 --- a/infrahub_sdk/constants.py +++ b/infrahub_sdk/constants.py @@ -1,5 +1,23 @@ import enum +# Namespaces reserved by the Infrahub server — mirrored from +# backend/infrahub/core/constants/__init__.py in the opsmill/infrahub repo. +# The server should eventually consume this list from the SDK. +RESTRICTED_NAMESPACES: list[str] = [ + "Account", + "Branch", + "Builtin", + "Core", + "Deprecated", + "Diff", + "Infrahub", + "Internal", + "Lineage", + "Schema", + "Profile", + "Template", +] + class InfrahubClientMode(str, enum.Enum): DEFAULT = "default" diff --git a/infrahub_sdk/ctl/schema.py b/infrahub_sdk/ctl/schema.py index 5a977b59..45cd7561 100644 --- a/infrahub_sdk/ctl/schema.py +++ b/infrahub_sdk/ctl/schema.py @@ -2,6 +2,7 @@ import asyncio import time +from datetime import datetime, timezone from pathlib import Path from typing import TYPE_CHECKING, Any @@ -11,10 +12,11 @@ from rich.console import Console from ..async_typer import AsyncTyper +from ..constants import RESTRICTED_NAMESPACES from ..ctl.client import initialize_client from ..ctl.utils import catch_exception, init_logging from ..queries import SCHEMA_HASH_SYNC_STATUS -from ..schema import SchemaWarning +from ..schema import GenericSchemaAPI, NodeSchemaAPI, ProfileSchemaAPI, SchemaWarning, TemplateSchemaAPI from ..yaml import SchemaFile from .parameters import CONFIG_PARAM from .utils import load_yamlfile_from_disk_and_exit @@ -211,3 +213,89 @@ def _display_schema_warnings(console: Console, warnings: list[SchemaWarning]) -> console.print( f"[yellow] {warning.type.value}: {warning.message} [{', '.join([kind.display for kind in warning.kinds])}]" ) + + +def _default_export_directory() -> str: + timestamp = datetime.now(timezone.utc).astimezone().strftime("%Y%m%d-%H%M%S") + return f"infrahub-schema-export-{timestamp}" + + +_SCHEMA_EXPORT_EXCLUDE: set[str] = {"hash", "hierarchy", "used_by", "id", "state"} +_FIELD_EXPORT_EXCLUDE: set[str] = {"inherited", "read_only", "allow_override", "hierarchical", "id", "state"} + + +def _schema_to_export_dict(schema: NodeSchemaAPI | GenericSchemaAPI) -> dict[str, Any]: + """Convert an API schema object to an export-ready dict (omits API-internal fields).""" + data = schema.model_dump(exclude=_SCHEMA_EXPORT_EXCLUDE, exclude_none=True) + + data["attributes"] = [ + dict(attr.model_dump(exclude=_FIELD_EXPORT_EXCLUDE, exclude_none=True)) + for attr in schema.attributes + if not attr.inherited + ] + if not data["attributes"]: + data.pop("attributes") + + data["relationships"] = [ + dict(rel.model_dump(exclude=_FIELD_EXPORT_EXCLUDE, exclude_none=True)) + for rel in schema.relationships + if not rel.inherited + ] + if not data["relationships"]: + data.pop("relationships") + + return data + + +@app.command() +@catch_exception(console=console) +async def export( + directory: Path = typer.Option(_default_export_directory, help="Directory path to store schema files"), + branch: str = typer.Option(None, help="Branch from which to export the schema"), + namespace: list[str] = typer.Option([], help="Namespace(s) to export (default: all user-defined)"), + debug: bool = False, + _: str = CONFIG_PARAM, +) -> None: + """Export the schema from Infrahub as YAML files, one per namespace.""" + init_logging(debug=debug) + + client = initialize_client() + schema_nodes = await client.schema.fetch(branch=branch or client.default_branch) + + user_schemas: dict[str, dict[str, list[dict[str, Any]]]] = {} + for schema in schema_nodes.values(): + if isinstance(schema, (ProfileSchemaAPI, TemplateSchemaAPI)): + continue + if schema.namespace in RESTRICTED_NAMESPACES: + continue + if namespace and schema.namespace not in namespace: + continue + ns = schema.namespace + user_schemas.setdefault(ns, {"nodes": [], "generics": []}) + schema_dict = _schema_to_export_dict(schema) + if isinstance(schema, GenericSchemaAPI): + user_schemas[ns]["generics"].append(schema_dict) + else: + user_schemas[ns]["nodes"].append(schema_dict) + + if not user_schemas: + console.print("[yellow]No user-defined schema found to export.") + return + + directory.mkdir(parents=True, exist_ok=True) + + for ns, data in sorted(user_schemas.items()): + payload: dict[str, Any] = {"version": "1.0"} + if data["nodes"]: + payload["nodes"] = data["nodes"] + if data["generics"]: + payload["generics"] = data["generics"] + + output_file = directory / f"{ns.lower()}.yml" + output_file.write_text( + yaml.dump(payload, default_flow_style=False, sort_keys=False, allow_unicode=True), + encoding="utf-8", + ) + console.print(f"[green] Exported namespace '{ns}' to {output_file}") + + console.print(f"[green] Schema exported to {directory}") diff --git a/tests/unit/ctl/test_schema_export.py b/tests/unit/ctl/test_schema_export.py new file mode 100644 index 00000000..f32b2413 --- /dev/null +++ b/tests/unit/ctl/test_schema_export.py @@ -0,0 +1,288 @@ +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +import yaml +from typer.testing import CliRunner + +if TYPE_CHECKING: + from pytest_httpx import HTTPXMock + +from infrahub_sdk.ctl.schema import app +from tests.helpers.cli import remove_ansi_color + +runner = CliRunner() + +# --------------------------------------------------------------------------- +# Minimal schema API response builders +# --------------------------------------------------------------------------- + +_BASE_NODE = { + "id": None, + "state": "present", + "hash": None, + "hierarchy": None, + "label": None, + "description": None, + "include_in_menu": None, + "menu_placement": None, + "display_label": None, + "display_labels": None, + "human_friendly_id": None, + "icon": None, + "uniqueness_constraints": None, + "documentation": None, + "order_by": None, + "inherit_from": [], + "branch": "aware", + "default_filter": None, + "generate_profile": None, + "generate_template": None, + "parent": None, + "children": None, + "attributes": [], + "relationships": [], +} + +_BASE_GENERIC = { + "id": None, + "state": "present", + "hash": None, + "used_by": [], + "label": None, + "description": None, + "include_in_menu": None, + "menu_placement": None, + "display_label": None, + "display_labels": None, + "human_friendly_id": None, + "icon": None, + "uniqueness_constraints": None, + "documentation": None, + "order_by": None, + "attributes": [], + "relationships": [], +} + + +def _make_node(namespace: str, name: str, **kwargs: object) -> dict: + node = {**_BASE_NODE, "namespace": namespace, "name": name} + node.update(kwargs) + return node + + +def _make_generic(namespace: str, name: str, **kwargs: object) -> dict: + generic = {**_BASE_GENERIC, "namespace": namespace, "name": name} + generic.update(kwargs) + return generic + + +def _make_profile(namespace: str, name: str) -> dict: + return { + "id": None, + "state": "present", + "namespace": namespace, + "name": name, + "label": None, + "description": None, + "include_in_menu": None, + "menu_placement": None, + "display_label": None, + "display_labels": None, + "human_friendly_id": None, + "icon": None, + "uniqueness_constraints": None, + "documentation": None, + "order_by": None, + "inherit_from": [], + "attributes": [], + "relationships": [], + } + + +def _schema_response( + nodes: list[dict] | None = None, + generics: list[dict] | None = None, + profiles: list[dict] | None = None, + templates: list[dict] | None = None, +) -> dict: + return { + "main": "aabbccdd", + "nodes": nodes or [], + "generics": generics or [], + "profiles": profiles or [], + "templates": templates or [], + } + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +def test_schema_export_basic(httpx_mock: HTTPXMock, tmp_path: Path) -> None: + """Two user namespaces produce two YAML files with correct content.""" + response = _schema_response( + nodes=[ + _make_node("Infra", "Device"), + _make_node("Dcim", "Rack"), + ] + ) + httpx_mock.add_response( + method="GET", + url="http://mock/api/schema?branch=main", + json=response, + ) + + output_dir = tmp_path / "export" + result = runner.invoke(app=app, args=["export", "--directory", str(output_dir)]) + + assert result.exit_code == 0, result.stdout + clean = remove_ansi_color(result.stdout) + assert "Exported namespace 'Dcim'" in clean + assert "Exported namespace 'Infra'" in clean + + dcim_file = output_dir / "dcim.yml" + infra_file = output_dir / "infra.yml" + assert dcim_file.exists() + assert infra_file.exists() + + dcim_data = yaml.safe_load(dcim_file.read_text()) + infra_data = yaml.safe_load(infra_file.read_text()) + + assert dcim_data["version"] == "1.0" + assert any(n["name"] == "Rack" for n in dcim_data["nodes"]) + + assert infra_data["version"] == "1.0" + assert any(n["name"] == "Device" for n in infra_data["nodes"]) + + +def test_schema_export_excludes_restricted_namespaces(httpx_mock: HTTPXMock, tmp_path: Path) -> None: + """Restricted namespaces (Core, Builtin, Internal, etc.) are excluded from export.""" + response = _schema_response( + nodes=[ + _make_node("Core", "MenuItem"), + _make_node("Builtin", "Tag"), + _make_node("Internal", "Node"), + ] + ) + httpx_mock.add_response( + method="GET", + url="http://mock/api/schema?branch=main", + json=response, + ) + + output_dir = tmp_path / "export" + result = runner.invoke(app=app, args=["export", "--directory", str(output_dir)]) + + assert result.exit_code == 0, result.stdout + assert "No user-defined schema found" in remove_ansi_color(result.stdout) + assert not output_dir.exists() + + +def test_schema_export_excludes_profiles_templates(httpx_mock: HTTPXMock, tmp_path: Path) -> None: + """ProfileSchemaAPI and TemplateSchemaAPI objects are skipped.""" + profile = _make_profile("Infra", "ProfileDevice") + template = {**profile, "name": "TemplateDevice"} + response = _schema_response(profiles=[profile], templates=[template]) + httpx_mock.add_response( + method="GET", + url="http://mock/api/schema?branch=main", + json=response, + ) + + output_dir = tmp_path / "export" + result = runner.invoke(app=app, args=["export", "--directory", str(output_dir)]) + + assert result.exit_code == 0, result.stdout + assert "No user-defined schema found" in remove_ansi_color(result.stdout) + assert not output_dir.exists() + + +def test_schema_export_namespace_filter(httpx_mock: HTTPXMock, tmp_path: Path) -> None: + """--namespace flag limits export to the specified namespace.""" + response = _schema_response( + nodes=[ + _make_node("Infra", "Device"), + _make_node("Dcim", "Rack"), + ] + ) + httpx_mock.add_response( + method="GET", + url="http://mock/api/schema?branch=main", + json=response, + ) + + output_dir = tmp_path / "export" + result = runner.invoke(app=app, args=["export", "--directory", str(output_dir), "--namespace", "Infra"]) + + assert result.exit_code == 0, result.stdout + assert (output_dir / "infra.yml").exists() + assert not (output_dir / "dcim.yml").exists() + + +def test_schema_export_no_user_schema(httpx_mock: HTTPXMock, tmp_path: Path) -> None: + """No output files when all schemas are in restricted namespaces.""" + response = _schema_response( + nodes=[ + _make_node("Core", "Repository"), + _make_node("Builtin", "Tag"), + _make_node("Internal", "Node"), + ] + ) + httpx_mock.add_response( + method="GET", + url="http://mock/api/schema?branch=main", + json=response, + ) + + output_dir = tmp_path / "export" + result = runner.invoke(app=app, args=["export", "--directory", str(output_dir)]) + + assert result.exit_code == 0, result.stdout + assert "No user-defined schema found" in remove_ansi_color(result.stdout) + assert not output_dir.exists() + + +def test_schema_export_custom_directory(httpx_mock: HTTPXMock, tmp_path: Path) -> None: + """Files are created in the directory specified via --directory.""" + response = _schema_response(nodes=[_make_node("Network", "Prefix")]) + httpx_mock.add_response( + method="GET", + url="http://mock/api/schema?branch=main", + json=response, + ) + + custom_dir = tmp_path / "my-custom-export" + result = runner.invoke(app=app, args=["export", "--directory", str(custom_dir)]) + + assert result.exit_code == 0, result.stdout + assert (custom_dir / "network.yml").exists() + clean = remove_ansi_color(result.stdout) + assert "Schema exported to" in clean + assert custom_dir.name in clean + + +def test_schema_export_includes_generics(httpx_mock: HTTPXMock, tmp_path: Path) -> None: + """Generic schemas are exported under the 'generics' key.""" + response = _schema_response( + generics=[_make_generic("Infra", "GenericInterface")], + nodes=[_make_node("Infra", "Device")], + ) + httpx_mock.add_response( + method="GET", + url="http://mock/api/schema?branch=main", + json=response, + ) + + output_dir = tmp_path / "export" + result = runner.invoke(app=app, args=["export", "--directory", str(output_dir)]) + + assert result.exit_code == 0, result.stdout + infra_file = output_dir / "infra.yml" + assert infra_file.exists() + + data = yaml.safe_load(infra_file.read_text()) + assert any(g["name"] == "GenericInterface" for g in data["generics"]) + assert any(n["name"] == "Device" for n in data["nodes"]) From a9ecccd92ac56909553bc85d962c620825ce0984 Mon Sep 17 00:00:00 2001 From: Bearchitek Date: Thu, 19 Feb 2026 09:52:28 +0100 Subject: [PATCH 2/6] Add doc regeneration rule to AGENTS.md Remind agents to run `invoke generate-sdk generate-infrahubctl` after changing CLI commands or SDK config so generated docs stay in sync. Co-Authored-By: Claude Opus 4.6 --- AGENTS.md | 1 + 1 file changed, 1 insertion(+) diff --git a/AGENTS.md b/AGENTS.md index 576ea00e..00de5ab1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -54,6 +54,7 @@ Key rules: ✅ **Always** - Run `uv run invoke format lint-code` before committing Python code +- Run `uv run invoke generate-sdk generate-infrahubctl` after changing CLI commands or SDK config - Run markdownlint before committing markdown changes - Follow async/sync dual pattern for new features - Use type hints on all function signatures From 7d70cb06314644d73c15fcbcc92c0a270c0e282f Mon Sep 17 00:00:00 2001 From: Benoit Kohler Date: Thu, 19 Feb 2026 16:56:24 +0100 Subject: [PATCH 3/6] Update constants.py --- infrahub_sdk/constants.py | 19 ------------------- 1 file changed, 19 deletions(-) diff --git a/infrahub_sdk/constants.py b/infrahub_sdk/constants.py index b3a53783..4eee947a 100644 --- a/infrahub_sdk/constants.py +++ b/infrahub_sdk/constants.py @@ -1,24 +1,5 @@ import enum -# Namespaces reserved by the Infrahub server — mirrored from -# backend/infrahub/core/constants/__init__.py in the opsmill/infrahub repo. -# The server should eventually consume this list from the SDK. -RESTRICTED_NAMESPACES: list[str] = [ - "Account", - "Branch", - "Builtin", - "Core", - "Deprecated", - "Diff", - "Infrahub", - "Internal", - "Lineage", - "Schema", - "Profile", - "Template", -] - - class InfrahubClientMode(str, enum.Enum): DEFAULT = "default" TRACKING = "tracking" From 7d2f7a012459e1640b9b293af7c7f3384ba6341a Mon Sep 17 00:00:00 2001 From: Benoit Kohler Date: Thu, 19 Feb 2026 16:56:40 +0100 Subject: [PATCH 4/6] Update constants.py --- infrahub_sdk/constants.py | 1 + 1 file changed, 1 insertion(+) diff --git a/infrahub_sdk/constants.py b/infrahub_sdk/constants.py index 4eee947a..04dd6b95 100644 --- a/infrahub_sdk/constants.py +++ b/infrahub_sdk/constants.py @@ -1,5 +1,6 @@ import enum + class InfrahubClientMode(str, enum.Enum): DEFAULT = "default" TRACKING = "tracking" From e38de3bbfa7eaa3c4c24c7e12ceb42661e90e028 Mon Sep 17 00:00:00 2001 From: Benoit Kohler Date: Thu, 19 Feb 2026 16:57:46 +0100 Subject: [PATCH 5/6] Update schema.py --- infrahub_sdk/ctl/schema.py | 90 +------------------------------------- 1 file changed, 1 insertion(+), 89 deletions(-) diff --git a/infrahub_sdk/ctl/schema.py b/infrahub_sdk/ctl/schema.py index 45cd7561..5a977b59 100644 --- a/infrahub_sdk/ctl/schema.py +++ b/infrahub_sdk/ctl/schema.py @@ -2,7 +2,6 @@ import asyncio import time -from datetime import datetime, timezone from pathlib import Path from typing import TYPE_CHECKING, Any @@ -12,11 +11,10 @@ from rich.console import Console from ..async_typer import AsyncTyper -from ..constants import RESTRICTED_NAMESPACES from ..ctl.client import initialize_client from ..ctl.utils import catch_exception, init_logging from ..queries import SCHEMA_HASH_SYNC_STATUS -from ..schema import GenericSchemaAPI, NodeSchemaAPI, ProfileSchemaAPI, SchemaWarning, TemplateSchemaAPI +from ..schema import SchemaWarning from ..yaml import SchemaFile from .parameters import CONFIG_PARAM from .utils import load_yamlfile_from_disk_and_exit @@ -213,89 +211,3 @@ def _display_schema_warnings(console: Console, warnings: list[SchemaWarning]) -> console.print( f"[yellow] {warning.type.value}: {warning.message} [{', '.join([kind.display for kind in warning.kinds])}]" ) - - -def _default_export_directory() -> str: - timestamp = datetime.now(timezone.utc).astimezone().strftime("%Y%m%d-%H%M%S") - return f"infrahub-schema-export-{timestamp}" - - -_SCHEMA_EXPORT_EXCLUDE: set[str] = {"hash", "hierarchy", "used_by", "id", "state"} -_FIELD_EXPORT_EXCLUDE: set[str] = {"inherited", "read_only", "allow_override", "hierarchical", "id", "state"} - - -def _schema_to_export_dict(schema: NodeSchemaAPI | GenericSchemaAPI) -> dict[str, Any]: - """Convert an API schema object to an export-ready dict (omits API-internal fields).""" - data = schema.model_dump(exclude=_SCHEMA_EXPORT_EXCLUDE, exclude_none=True) - - data["attributes"] = [ - dict(attr.model_dump(exclude=_FIELD_EXPORT_EXCLUDE, exclude_none=True)) - for attr in schema.attributes - if not attr.inherited - ] - if not data["attributes"]: - data.pop("attributes") - - data["relationships"] = [ - dict(rel.model_dump(exclude=_FIELD_EXPORT_EXCLUDE, exclude_none=True)) - for rel in schema.relationships - if not rel.inherited - ] - if not data["relationships"]: - data.pop("relationships") - - return data - - -@app.command() -@catch_exception(console=console) -async def export( - directory: Path = typer.Option(_default_export_directory, help="Directory path to store schema files"), - branch: str = typer.Option(None, help="Branch from which to export the schema"), - namespace: list[str] = typer.Option([], help="Namespace(s) to export (default: all user-defined)"), - debug: bool = False, - _: str = CONFIG_PARAM, -) -> None: - """Export the schema from Infrahub as YAML files, one per namespace.""" - init_logging(debug=debug) - - client = initialize_client() - schema_nodes = await client.schema.fetch(branch=branch or client.default_branch) - - user_schemas: dict[str, dict[str, list[dict[str, Any]]]] = {} - for schema in schema_nodes.values(): - if isinstance(schema, (ProfileSchemaAPI, TemplateSchemaAPI)): - continue - if schema.namespace in RESTRICTED_NAMESPACES: - continue - if namespace and schema.namespace not in namespace: - continue - ns = schema.namespace - user_schemas.setdefault(ns, {"nodes": [], "generics": []}) - schema_dict = _schema_to_export_dict(schema) - if isinstance(schema, GenericSchemaAPI): - user_schemas[ns]["generics"].append(schema_dict) - else: - user_schemas[ns]["nodes"].append(schema_dict) - - if not user_schemas: - console.print("[yellow]No user-defined schema found to export.") - return - - directory.mkdir(parents=True, exist_ok=True) - - for ns, data in sorted(user_schemas.items()): - payload: dict[str, Any] = {"version": "1.0"} - if data["nodes"]: - payload["nodes"] = data["nodes"] - if data["generics"]: - payload["generics"] = data["generics"] - - output_file = directory / f"{ns.lower()}.yml" - output_file.write_text( - yaml.dump(payload, default_flow_style=False, sort_keys=False, allow_unicode=True), - encoding="utf-8", - ) - console.print(f"[green] Exported namespace '{ns}' to {output_file}") - - console.print(f"[green] Schema exported to {directory}") From 45ecc8fc5688ba5b7c7f98adc000681f59f88f53 Mon Sep 17 00:00:00 2001 From: Benoit Kohler Date: Thu, 19 Feb 2026 16:58:01 +0100 Subject: [PATCH 6/6] Delete tests/unit/ctl/test_schema_export.py --- tests/unit/ctl/test_schema_export.py | 288 --------------------------- 1 file changed, 288 deletions(-) delete mode 100644 tests/unit/ctl/test_schema_export.py diff --git a/tests/unit/ctl/test_schema_export.py b/tests/unit/ctl/test_schema_export.py deleted file mode 100644 index f32b2413..00000000 --- a/tests/unit/ctl/test_schema_export.py +++ /dev/null @@ -1,288 +0,0 @@ -from __future__ import annotations - -from pathlib import Path -from typing import TYPE_CHECKING - -import yaml -from typer.testing import CliRunner - -if TYPE_CHECKING: - from pytest_httpx import HTTPXMock - -from infrahub_sdk.ctl.schema import app -from tests.helpers.cli import remove_ansi_color - -runner = CliRunner() - -# --------------------------------------------------------------------------- -# Minimal schema API response builders -# --------------------------------------------------------------------------- - -_BASE_NODE = { - "id": None, - "state": "present", - "hash": None, - "hierarchy": None, - "label": None, - "description": None, - "include_in_menu": None, - "menu_placement": None, - "display_label": None, - "display_labels": None, - "human_friendly_id": None, - "icon": None, - "uniqueness_constraints": None, - "documentation": None, - "order_by": None, - "inherit_from": [], - "branch": "aware", - "default_filter": None, - "generate_profile": None, - "generate_template": None, - "parent": None, - "children": None, - "attributes": [], - "relationships": [], -} - -_BASE_GENERIC = { - "id": None, - "state": "present", - "hash": None, - "used_by": [], - "label": None, - "description": None, - "include_in_menu": None, - "menu_placement": None, - "display_label": None, - "display_labels": None, - "human_friendly_id": None, - "icon": None, - "uniqueness_constraints": None, - "documentation": None, - "order_by": None, - "attributes": [], - "relationships": [], -} - - -def _make_node(namespace: str, name: str, **kwargs: object) -> dict: - node = {**_BASE_NODE, "namespace": namespace, "name": name} - node.update(kwargs) - return node - - -def _make_generic(namespace: str, name: str, **kwargs: object) -> dict: - generic = {**_BASE_GENERIC, "namespace": namespace, "name": name} - generic.update(kwargs) - return generic - - -def _make_profile(namespace: str, name: str) -> dict: - return { - "id": None, - "state": "present", - "namespace": namespace, - "name": name, - "label": None, - "description": None, - "include_in_menu": None, - "menu_placement": None, - "display_label": None, - "display_labels": None, - "human_friendly_id": None, - "icon": None, - "uniqueness_constraints": None, - "documentation": None, - "order_by": None, - "inherit_from": [], - "attributes": [], - "relationships": [], - } - - -def _schema_response( - nodes: list[dict] | None = None, - generics: list[dict] | None = None, - profiles: list[dict] | None = None, - templates: list[dict] | None = None, -) -> dict: - return { - "main": "aabbccdd", - "nodes": nodes or [], - "generics": generics or [], - "profiles": profiles or [], - "templates": templates or [], - } - - -# --------------------------------------------------------------------------- -# Tests -# --------------------------------------------------------------------------- - - -def test_schema_export_basic(httpx_mock: HTTPXMock, tmp_path: Path) -> None: - """Two user namespaces produce two YAML files with correct content.""" - response = _schema_response( - nodes=[ - _make_node("Infra", "Device"), - _make_node("Dcim", "Rack"), - ] - ) - httpx_mock.add_response( - method="GET", - url="http://mock/api/schema?branch=main", - json=response, - ) - - output_dir = tmp_path / "export" - result = runner.invoke(app=app, args=["export", "--directory", str(output_dir)]) - - assert result.exit_code == 0, result.stdout - clean = remove_ansi_color(result.stdout) - assert "Exported namespace 'Dcim'" in clean - assert "Exported namespace 'Infra'" in clean - - dcim_file = output_dir / "dcim.yml" - infra_file = output_dir / "infra.yml" - assert dcim_file.exists() - assert infra_file.exists() - - dcim_data = yaml.safe_load(dcim_file.read_text()) - infra_data = yaml.safe_load(infra_file.read_text()) - - assert dcim_data["version"] == "1.0" - assert any(n["name"] == "Rack" for n in dcim_data["nodes"]) - - assert infra_data["version"] == "1.0" - assert any(n["name"] == "Device" for n in infra_data["nodes"]) - - -def test_schema_export_excludes_restricted_namespaces(httpx_mock: HTTPXMock, tmp_path: Path) -> None: - """Restricted namespaces (Core, Builtin, Internal, etc.) are excluded from export.""" - response = _schema_response( - nodes=[ - _make_node("Core", "MenuItem"), - _make_node("Builtin", "Tag"), - _make_node("Internal", "Node"), - ] - ) - httpx_mock.add_response( - method="GET", - url="http://mock/api/schema?branch=main", - json=response, - ) - - output_dir = tmp_path / "export" - result = runner.invoke(app=app, args=["export", "--directory", str(output_dir)]) - - assert result.exit_code == 0, result.stdout - assert "No user-defined schema found" in remove_ansi_color(result.stdout) - assert not output_dir.exists() - - -def test_schema_export_excludes_profiles_templates(httpx_mock: HTTPXMock, tmp_path: Path) -> None: - """ProfileSchemaAPI and TemplateSchemaAPI objects are skipped.""" - profile = _make_profile("Infra", "ProfileDevice") - template = {**profile, "name": "TemplateDevice"} - response = _schema_response(profiles=[profile], templates=[template]) - httpx_mock.add_response( - method="GET", - url="http://mock/api/schema?branch=main", - json=response, - ) - - output_dir = tmp_path / "export" - result = runner.invoke(app=app, args=["export", "--directory", str(output_dir)]) - - assert result.exit_code == 0, result.stdout - assert "No user-defined schema found" in remove_ansi_color(result.stdout) - assert not output_dir.exists() - - -def test_schema_export_namespace_filter(httpx_mock: HTTPXMock, tmp_path: Path) -> None: - """--namespace flag limits export to the specified namespace.""" - response = _schema_response( - nodes=[ - _make_node("Infra", "Device"), - _make_node("Dcim", "Rack"), - ] - ) - httpx_mock.add_response( - method="GET", - url="http://mock/api/schema?branch=main", - json=response, - ) - - output_dir = tmp_path / "export" - result = runner.invoke(app=app, args=["export", "--directory", str(output_dir), "--namespace", "Infra"]) - - assert result.exit_code == 0, result.stdout - assert (output_dir / "infra.yml").exists() - assert not (output_dir / "dcim.yml").exists() - - -def test_schema_export_no_user_schema(httpx_mock: HTTPXMock, tmp_path: Path) -> None: - """No output files when all schemas are in restricted namespaces.""" - response = _schema_response( - nodes=[ - _make_node("Core", "Repository"), - _make_node("Builtin", "Tag"), - _make_node("Internal", "Node"), - ] - ) - httpx_mock.add_response( - method="GET", - url="http://mock/api/schema?branch=main", - json=response, - ) - - output_dir = tmp_path / "export" - result = runner.invoke(app=app, args=["export", "--directory", str(output_dir)]) - - assert result.exit_code == 0, result.stdout - assert "No user-defined schema found" in remove_ansi_color(result.stdout) - assert not output_dir.exists() - - -def test_schema_export_custom_directory(httpx_mock: HTTPXMock, tmp_path: Path) -> None: - """Files are created in the directory specified via --directory.""" - response = _schema_response(nodes=[_make_node("Network", "Prefix")]) - httpx_mock.add_response( - method="GET", - url="http://mock/api/schema?branch=main", - json=response, - ) - - custom_dir = tmp_path / "my-custom-export" - result = runner.invoke(app=app, args=["export", "--directory", str(custom_dir)]) - - assert result.exit_code == 0, result.stdout - assert (custom_dir / "network.yml").exists() - clean = remove_ansi_color(result.stdout) - assert "Schema exported to" in clean - assert custom_dir.name in clean - - -def test_schema_export_includes_generics(httpx_mock: HTTPXMock, tmp_path: Path) -> None: - """Generic schemas are exported under the 'generics' key.""" - response = _schema_response( - generics=[_make_generic("Infra", "GenericInterface")], - nodes=[_make_node("Infra", "Device")], - ) - httpx_mock.add_response( - method="GET", - url="http://mock/api/schema?branch=main", - json=response, - ) - - output_dir = tmp_path / "export" - result = runner.invoke(app=app, args=["export", "--directory", str(output_dir)]) - - assert result.exit_code == 0, result.stdout - infra_file = output_dir / "infra.yml" - assert infra_file.exists() - - data = yaml.safe_load(infra_file.read_text()) - assert any(g["name"] == "GenericInterface" for g in data["generics"]) - assert any(n["name"] == "Device" for n in data["nodes"])