feat: IHS-190 (part of IFC-2162) Add namespace restriction parameter to generic schemas

infrahub_sdk/schema/main.py

@@ -289,7 +289,9 @@ class GenericSchemaAPI(BaseSchema, BaseSchemaAttrRelAPI):
     """A Generic can be either an Interface or a Union depending if there are some Attributes or Relationships defined."""
 
     hash: str | None = None
+    hierarchical: bool | None = None
     used_by: list[str] = Field(default_factory=list)
+    restricted_namespaces: list[str] | None = None
 
 
 class BaseNodeSchema(BaseSchema):

tests/fixtures/models/valid_generic_schema.json

@@ -0,0 +1,30 @@
+{
+    "version": "1.0",
+    "generics": [
+        {
+            "name": "Animal",
+            "namespace": "Testing",
+            "attributes": [
+                {
+                    "name": "name",
+                    "kind": "Text"
+                }
+            ],
+            "restricted_namespaces": ["Dog"]
+        }
+    ],
+    "nodes": [
+        {
+            "name": "Dog",
+            "namespace": "Dog",
+            "inherit_from": ["TestingAnimal"],
+            "attributes": [
+                {
+                    "name": "breed",
+                    "kind": "Text",
+                    "optional": true
+                }
+            ]
+        }
+    ]
+}

tests/unit/ctl/test_schema_app.py

@@ -128,3 +128,48 @@ def test_schema_load_notvalid_namespace(httpx_mock: HTTPXMock) -> None:
         fixture_file.read_text(encoding="utf-8"),
     )
     assert content_json == {"schemas": [fixture_file_content]}
+
+
+def test_load_valid_generic_schema(httpx_mock: HTTPXMock) -> None:
+    """A test which ensures that a generic schema is correctly loaded when loaded from infrahubctl command"""
+
+    # Arrange
+    fixture_file = get_fixtures_dir() / "models" / "valid_generic_schema.json"
+
+    httpx_mock.add_response(
+        method="POST",
+        url="http://mock/api/schema/load?branch=main",
+        status_code=200,
+        json={
+            "hash": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4",
+            "previous_hash": "d3f7f4e7161f0ae6538a01d5a42dc661",
+            "diff": {
+                "added": {
+                    "TestingAnimal": {"added": {}, "changed": {}, "removed": {}},
+                    "DogDog": {"added": {}, "changed": {}, "removed": {}},
+                },
+                "changed": {},
+                "removed": {},
+            },
+            "schema_updated": True,
+        },
+    )
+
+    # Act
+    result = runner.invoke(app=app, args=["load", str(fixture_file)])
+
+    # Assert
+    assert result.exit_code == 0
+    assert f"schema '{fixture_file}' loaded successfully" in remove_ansi_color(result.stdout.replace("\n", ""))
+
+    content = httpx_mock.get_requests()[0].content.decode("utf8")
+    content_json = yaml.safe_load(content)
+    fixture_file_content = yaml.safe_load(
+        fixture_file.read_text(encoding="utf-8"),
+    )
+    assert content_json == {"schemas": [fixture_file_content]}
+
+    # Verify restricted_namespaces is present in the payload sent to the API
+    sent_generics = content_json["schemas"][0]["generics"]
+    assert len(sent_generics) == 1
+    assert sent_generics[0]["restricted_namespaces"] == ["Dog"]

tests/unit/sdk/test_schema.py

@@ -482,3 +482,28 @@ def test_schema_base__get_schema_name__returns_correct_schema_name_for_protocols
     assert InfrahubSchemaBase._get_schema_name(schema=BuiltinIPAddressSync) == "BuiltinIPAddress"
     assert InfrahubSchemaBase._get_schema_name(schema=BuiltinIPAddress) == "BuiltinIPAddress"
     assert InfrahubSchemaBase._get_schema_name(schema="BuiltinIPAddress") == "BuiltinIPAddress"
+
+
+async def test_schema_load_surfaces_api_error_on_422(client: InfrahubClient, httpx_mock: HTTPXMock) -> None:
+    """Validate that schema.load surfaces API error responses to the caller."""
+    # Arrange
+    schema_payload = {"version": "1.0", "nodes": [{"name": "Dummy", "namespace": "Test"}]}
+    api_error_message = "Something went wrong on the server side."
+
+    httpx_mock.add_response(
+        method="POST",
+        url="http://mock/api/schema/load?branch=main",
+        status_code=422,
+        json={
+            "data": None,
+            "errors": [{"message": api_error_message, "extensions": {"code": 422}}],
+        },
+    )
+
+    # Act
+    response = await client.schema.load(schemas=[schema_payload])
+
+    # Assert
+    assert response.errors
+    assert response.errors["errors"][0]["message"] == api_error_message
+    assert not response.schema_updated