Add method to validate format of object file

Author: dgarros

infrahub_sdk/ctl/object.py

@@ -40,8 +40,28 @@ async def load(
     client = initialize_client()
 
     for file in files:
-        file.validate_content()
-        schema = await client.schema.get(kind=file.spec.kind, branch=branch)
+        await file.validate_format(client=client, branch=branch)
 
-        for item in file.spec.data:
-            await file.spec.create_node(client=client, schema=schema, data=item, branch=branch)
+    for file in files:
+        await file.process(client=client, branch=branch)
+
+
+@app.command()
+@catch_exception(console=console)
+async def validate(
+    paths: list[Path],
+    debug: bool = False,
+    branch: str = typer.Option(None, help="Branch on which to validate the objects."),
+    _: str = CONFIG_PARAM,
+) -> None:
+    """Validate one or multiple objects files."""
+
+    init_logging(debug=debug)
+
+    logging.getLogger("infrahub_sdk").setLevel(logging.INFO)
+
+    files = load_yamlfile_from_disk_and_exit(paths=paths, file_type=ObjectFile, console=console)
+    client = initialize_client()
+
+    for file in files:
+        await file.validate_format(client=client, branch=branch)

infrahub_sdk/schema/main.py

@@ -143,7 +143,6 @@ class RelationshipSchema(BaseModel):
 class RelationshipSchemaAPI(RelationshipSchema):
     model_config = ConfigDict(use_enum_values=True)
 
-    identifier: str
     inherited: bool = False
     read_only: bool = False
     hierarchical: str | None = None

infrahub_sdk/spec/object.py

@@ -4,6 +4,7 @@
 
 from pydantic import BaseModel, Field
 
+from ..exceptions import ValidationError
 from ..schema import RelationshipSchema
 from ..yaml import InfrahubFile, InfrahubFileKind
 
@@ -56,7 +57,9 @@ async def get_relationship_info(
     peer_schema = await client.schema.get(kind=info.peer_kind, branch=branch)
 
     try:
-        info.peer_rel = peer_schema.get_matching_relationship(id=rel_schema.identifier, direction=rel_schema.direction)
+        info.peer_rel = peer_schema.get_matching_relationship(
+            id=rel_schema.identifier or "", direction=rel_schema.direction
+        )
     except ValueError:
         pass
 
@@ -72,6 +75,64 @@ class InfrahubObjectFileData(BaseModel):
     kind: str
     data: list[dict[str, Any]] = Field(default_factory=list)
 
+    async def validate_format(self, client: InfrahubClient, branch: str | None = None) -> list[ValidationError]:
+        errors: list[ValidationError] = []
+        schema = await client.schema.get(kind=self.kind, branch=branch)
+        for item in self.data:
+            errors.extend(await self.validate_object(client=client, schema=schema, data=item, branch=branch))
+        return errors
+
+    async def process(self, client: InfrahubClient, branch: str | None = None) -> None:
+        schema = await client.schema.get(kind=self.kind, branch=branch)
+        for item in self.data:
+            await self.create_node(client=client, schema=schema, data=item, branch=branch)
+
+    @classmethod
+    async def validate_object(
+        cls,
+        client: InfrahubClient,
+        schema: MainSchemaTypesAPI,
+        data: dict,
+        context: dict | None = None,
+        branch: str | None = None,
+    ) -> list[ValidationError]:
+        errors: list[ValidationError] = []
+        context = context or {}
+
+        # First validate if all mandatory fields are present
+        for element in schema.mandatory_attribute_names + schema.mandatory_relationship_names:
+            if not any([element in data.keys(), element in context.keys()]):
+                errors.append(ValidationError(identifier=element, message=f"{element} is mandatory"))
+
+        # Validate if all attributes are valid
+        for key, value in data.items():
+            if key not in schema.attribute_names and key not in schema.relationship_names:
+                errors.append(
+                    ValidationError(identifier=key, message=f"{key} is not a valid attribute or relationship")
+                )
+
+            if key in schema.attribute_names:
+                if not isinstance(value, (str, int, float, bool, list, dict)):
+                    errors.append(
+                        ValidationError(
+                            identifier=key, message=f"{key} must be a string, int, float, bool, list, or dict"
+                        )
+                    )
+
+            if key in schema.relationship_names:
+                rel_info = await get_relationship_info(
+                    client=client, schema=schema, key=key, value=value, branch=branch
+                )
+                if not rel_info.is_valid:
+                    errors.append(
+                        ValidationError(
+                            identifier=key, message=rel_info.reason_relationship_not_valid or "Invalid relationship"
+                        )
+                    )
+                # TODO: Validate the sub-object too
+
+        return errors
+
     @classmethod
     def enrich_node(cls, data: dict, context: dict) -> dict:  # noqa: ARG003
         return data
@@ -88,10 +149,9 @@ async def create_node(
     ) -> InfrahubNode:
         context = context or {}
 
-        # First validate if all mandatory fields are present
-        for element in schema.mandatory_attribute_names + schema.mandatory_relationship_names:
-            if not any([element in data.keys(), element in context.keys()]):
-                raise ValueError(f"{element} is mandatory")
+        errors = await cls.validate_object(client=client, schema=schema, data=data, context=context, branch=branch)
+        if errors:
+            raise ValidationError(identifier=schema.kind, message="Object is not valid")
 
         clean_data: dict[str, Any] = {}
 
@@ -240,3 +300,13 @@ def validate_content(self) -> None:
         if self.kind != InfrahubFileKind.OBJECT:
             raise ValueError("File is not an Infrahub Object file")
         self._spec = InfrahubObjectFileData(**self.data.spec)
+
+    async def validate_format(self, client: InfrahubClient, branch: str | None = None) -> None:
+        self.validate_content()
+        errors = await self.spec.validate_format(client=client, branch=branch)
+        if errors:
+            error_message = "\n".join([f"{error.identifier} : {error.message}" for error in errors])
+            raise ValidationError(identifier=str(self.location), message=error_message)
+
+    async def process(self, client: InfrahubClient, branch: str | None = None) -> None:
+        await self.spec.process(client=client, branch=branch)

tests/integration/test_spec_object.py

@@ -10,7 +10,17 @@
 from infrahub_sdk.utils import get_fixtures_dir
 
 
+def load_object_file(name: str) -> ObjectFile:
+    files = ObjectFile.load_from_disk(paths=[get_fixtures_dir() / "spec_objects" / name])
+    assert len(files) == 1
+    return files[0]
+
+
 class TestSpecObject(TestInfrahubDockerClient, SchemaAnimal):
+    @pytest.fixture(scope="class")
+    def branch_name(self) -> str:
+        return "branch2"
+
     @pytest.fixture(scope="class")
     def spec_objects_fixtures_dir(self) -> Path:
         return get_fixtures_dir() / "spec_objects"
@@ -24,98 +34,66 @@ async def initial_schema(self, default_branch: str, client: InfrahubClient, sche
         )
         assert resp.errors == {}
 
-    async def test_load_tags(
-        self, client: InfrahubClient, default_branch: str, initial_schema: None, spec_objects_fixtures_dir: Path
-    ):
-        files = ObjectFile.load_from_disk(paths=[spec_objects_fixtures_dir / "animal_tags01.yml"])
-        assert len(files) == 1
-        obj_file = files[0]
+    async def test_create_branch(self, client: InfrahubClient, initial_schema: None, branch_name: str):
+        await client.branch.create(branch_name=branch_name, sync_with_git=False)
 
-        obj_file.validate_content()
+    async def test_load_tags(self, client: InfrahubClient, branch_name: str, initial_schema: None):
+        obj_file = load_object_file("animal_tags01.yml")
+        await obj_file.validate_format(client=client, branch=branch_name)
 
         # Check that the nodes are not present in the database before loading the file
-        assert len(await client.all(kind=obj_file.spec.kind)) == 0
+        assert len(await client.all(kind=obj_file.spec.kind, branch=branch_name)) == 0
 
-        schema = await client.schema.get(kind=obj_file.spec.kind, branch=default_branch)
-        for item in obj_file.spec.data:
-            await obj_file.spec.create_node(client=client, schema=schema, data=item, branch=default_branch)
+        await obj_file.process(client=client, branch=branch_name)
 
-        assert len(await client.all(kind=obj_file.spec.kind)) == 3
+        assert len(await client.all(kind=obj_file.spec.kind, branch=branch_name)) == 3
 
-    async def test_update_tags(
-        self, client: InfrahubClient, default_branch: str, initial_schema: None, spec_objects_fixtures_dir: Path
-    ):
-        files = ObjectFile.load_from_disk(paths=[spec_objects_fixtures_dir / "animal_tags02.yml"])
-        assert len(files) == 1
-        obj_file = files[0]
-
-        obj_file.validate_content()
+    async def test_update_tags(self, client: InfrahubClient, branch_name: str, initial_schema: None):
+        obj_file = load_object_file("animal_tags02.yml")
+        await obj_file.validate_format(client=client, branch=branch_name)
 
         # Check that the nodes are not present in the database before loading the file
-        assert len(await client.all(kind=obj_file.spec.kind)) == 3
+        assert len(await client.all(kind=obj_file.spec.kind, branch=branch_name)) == 3
 
-        schema = await client.schema.get(kind=obj_file.spec.kind, branch=default_branch)
-        for item in obj_file.spec.data:
-            await obj_file.spec.create_node(client=client, schema=schema, data=item, branch=default_branch)
+        await obj_file.process(client=client, branch=branch_name)
 
-        tags = await client.all(kind=obj_file.spec.kind)
+        tags = await client.all(kind=obj_file.spec.kind, branch=branch_name)
         tags_by_name = {"__".join(tag.get_human_friendly_id()): tag for tag in tags}
         assert len(tags_by_name) == 4
         assert tags_by_name["Veterinarian"].description.value == "Licensed animal healthcare professional"
 
-    async def test_load_persons(
-        self, client: InfrahubClient, default_branch: str, initial_schema: None, spec_objects_fixtures_dir: Path
-    ):
-        files = ObjectFile.load_from_disk(paths=[spec_objects_fixtures_dir / "animal_person01.yml"])
-        assert len(files) == 1
-        obj_file = files[0]
-
-        obj_file.validate_content()
+    async def test_load_persons(self, client: InfrahubClient, branch_name: str, initial_schema: None):
+        obj_file = load_object_file("animal_person01.yml")
+        await obj_file.validate_format(client=client, branch=branch_name)
 
         # Check that the nodes are not present in the database before loading the file
-        assert await client.all(kind=obj_file.spec.kind) == []
+        assert len(await client.all(kind=obj_file.spec.kind, branch=branch_name)) == 0
 
-        schema = await client.schema.get(kind=obj_file.spec.kind, branch=default_branch)
-        for item in obj_file.spec.data:
-            await obj_file.spec.create_node(client=client, schema=schema, data=item, branch=default_branch)
+        await obj_file.process(client=client, branch=branch_name)
 
-        assert len(await client.all(kind=obj_file.spec.kind)) == 3
+        assert len(await client.all(kind=obj_file.spec.kind, branch=branch_name)) == 3
 
-    async def test_load_dogs(
-        self, client: InfrahubClient, default_branch: str, initial_schema: None, spec_objects_fixtures_dir: Path
-    ):
-        files = ObjectFile.load_from_disk(paths=[spec_objects_fixtures_dir / "animal_dog01.yml"])
-        assert len(files) == 1
-        obj_file = files[0]
-
-        obj_file.validate_content()
+    async def test_load_dogs(self, client: InfrahubClient, branch_name: str, initial_schema: None):
+        obj_file = load_object_file("animal_dog01.yml")
+        await obj_file.validate_format(client=client, branch=branch_name)
 
         # Check that the nodes are not present in the database before loading the file
-        assert await client.all(kind=obj_file.spec.kind) == []
-
-        schema = await client.schema.get(kind=obj_file.spec.kind, branch=default_branch)
-        for item in obj_file.spec.data:
-            await obj_file.spec.create_node(client=client, schema=schema, data=item, branch=default_branch)
+        assert len(await client.all(kind=obj_file.spec.kind, branch=branch_name)) == 0
 
-        assert len(await client.all(kind=obj_file.spec.kind)) == 4
+        await obj_file.process(client=client, branch=branch_name)
 
-    async def test_load_persons02(
-        self, client: InfrahubClient, default_branch: str, initial_schema: None, spec_objects_fixtures_dir: Path
-    ):
-        files = ObjectFile.load_from_disk(paths=[spec_objects_fixtures_dir / "animal_person02.yml"])
-        assert len(files) == 1
-        obj_file = files[0]
+        assert len(await client.all(kind=obj_file.spec.kind, branch=branch_name)) == 4
 
-        obj_file.validate_content()
+    async def test_load_persons02(self, client: InfrahubClient, branch_name: str, initial_schema: None):
+        obj_file = load_object_file("animal_person02.yml")
+        await obj_file.validate_format(client=client, branch=branch_name)
 
         # Check that the nodes are not present in the database before loading the file
-        assert len(await client.all(kind=obj_file.spec.kind)) == 4
+        assert len(await client.all(kind=obj_file.spec.kind, branch=branch_name)) == 4
 
-        schema = await client.schema.get(kind=obj_file.spec.kind, branch=default_branch)
-        for item in obj_file.spec.data:
-            await obj_file.spec.create_node(client=client, schema=schema, data=item, branch=default_branch)
+        await obj_file.process(client=client, branch=branch_name)
 
-        persons = await client.all(kind=obj_file.spec.kind)
+        persons = await client.all(kind=obj_file.spec.kind, branch=branch_name)
         person_by_name = {"__".join(person.get_human_friendly_id()): person for person in persons}
         assert len(persons) == 5
 

tests/unit/sdk/spec/test_object.py

@@ -0,0 +1,63 @@
+import pytest
+from pytest_httpx import HTTPXMock
+
+from infrahub_sdk.client import InfrahubClient
+from infrahub_sdk.exceptions import ValidationError
+from infrahub_sdk.spec.object import ObjectFile
+
+
+@pytest.fixture
+def root_location() -> dict:
+    return {"apiVersion": "infrahub.app/v1", "kind": "Object", "spec": {"kind": "BuiltinLocation", "data": []}}
+
+
+@pytest.fixture
+def location_mexico_01(root_location: dict) -> dict:
+    data = [{"name": "Mexico", "type": "Country"}]
+
+    location = root_location.copy()
+    location["spec"]["data"] = data
+    return location
+
+
+@pytest.fixture
+def location_bad_syntax01(root_location: dict) -> dict:
+    data = [{"notthename": "Mexico", "type": "Country"}]
+    location = root_location.copy()
+    location["spec"]["data"] = data
+    return location
+
+
+@pytest.fixture
+def location_bad_syntax02(root_location: dict) -> dict:
+    data = [{"name": "Mexico", "notvalidattribute": "notvalidattribute", "type": "Country"}]
+    location = root_location.copy()
+    location["spec"]["data"] = data
+    return location
+
+
+async def test_validate_object(client: InfrahubClient, mock_schema_query_01: HTTPXMock, location_mexico_01):
+    obj = ObjectFile(location="some/path", content=location_mexico_01)
+    await obj.validate_format(client=client)
+
+    assert obj.spec.kind == "BuiltinLocation"
+
+
+async def test_validate_object_bad_syntax01(
+    client: InfrahubClient, mock_schema_query_01: HTTPXMock, location_bad_syntax01
+):
+    obj = ObjectFile(location="some/path", content=location_bad_syntax01)
+    with pytest.raises(ValidationError) as exc:
+        await obj.validate_format(client=client)
+
+    assert "name" in str(exc.value)
+
+
+async def test_validate_object_bad_syntax02(
+    client: InfrahubClient, mock_schema_query_01: HTTPXMock, location_bad_syntax02
+):
+    obj = ObjectFile(location="some/path", content=location_bad_syntax02)
+    with pytest.raises(ValidationError) as exc:
+        await obj.validate_format(client=client)
+
+    assert "notvalidattribute" in str(exc.value)