Add tests

Author: ogenstad

tests/fixtures/repos/ctl_integration/.infrahub.yml

@@ -0,0 +1,32 @@
+# yaml-language-server: $schema=https://schema.infrahub.app/python-sdk/repository-config/develop.json
+---
+python_transforms:
+  - name: animal_person
+    class_name: AnimalPerson
+    file_path: "transforms/animal_person.py"
+    convert_query_response: false
+  - name: animal_person_converted
+    class_name: ConvertedAnimalPerson
+    file_path: "transforms/converted.py"
+    convert_query_response: true
+
+generator_definitions:
+  - name: animal_tags
+    file_path: "generators/tag_generator.py"
+    targets: pet_owners
+    query: animal_person
+    convert_query_response: false
+    parameters:
+      name: "name__value"
+  - name: animal_tags_convert
+    file_path: "generators/tag_generator_convert.py"
+    targets: pet_owners
+    query: animal_person
+    convert_query_response: true
+    parameters:
+      name: "name__value"
+
+
+queries:
+  - name: animal_person
+    file_path: queries/animal_person.gql

tests/fixtures/repos/ctl_integration/generators/tag_generator.py

@@ -0,0 +1,15 @@
+from infrahub_sdk.generator import InfrahubGenerator
+
+
+class Generator(InfrahubGenerator):
+    async def generate(self, data: dict) -> None:
+        response_person = data["TestingPerson"]["edges"][0]["node"]
+        name: str = response_person["name"]["value"]
+
+        for animal in data["TestingPerson"]["edges"][0]["node"]["animals"]["edges"]:
+            payload = {
+                "name": f"raw-{name.lower().replace(' ', '-')}-{animal['node']['name']['value'].lower()}",
+                "description": "Without converting query response",
+            }
+            obj = await self.client.create(kind="BuiltinTag", data=payload)
+            await obj.save(allow_upsert=True)

tests/fixtures/repos/ctl_integration/generators/tag_generator_convert.py

@@ -0,0 +1,16 @@
+from infrahub_sdk.generator import InfrahubGenerator
+
+
+class Generator(InfrahubGenerator):
+    async def generate(self, data: dict) -> None:
+        response_person = data["TestingPerson"]["edges"][0]["node"]
+        name: str = response_person["name"]["value"]
+        person = self.store.get(key=name, kind="TestingPerson")
+
+        for animal in person.animals.peers:
+            payload = {
+                "name": f"converted-{name.lower().replace(' ', '-')}-{animal.peer.name.value.lower()}",
+                "description": "Using convert_query_response",
+            }
+            obj = await self.client.create(kind="BuiltinTag", data=payload)
+            await obj.save(allow_upsert=True)

tests/fixtures/repos/ctl_integration/queries/animal_person.gql

@@ -0,0 +1,27 @@
+query TestPersonQuery($name: String!) {
+  TestingPerson(name__value: $name) {
+    edges {
+      node {
+        __typename
+        id
+        name {
+          value
+        }
+        height {
+          value
+        }
+        animals {
+          edges {
+            node {
+              __typename
+              id
+              name {
+                value
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+}

tests/fixtures/repos/ctl_integration/transforms/animal_person.py

@@ -0,0 +1,16 @@
+from typing import Any
+
+from infrahub_sdk.transforms import InfrahubTransform
+
+
+class AnimalPerson(InfrahubTransform):
+    query = "animal_person"
+
+    async def transform(self, data) -> dict[str, Any]:
+        response_person = data["TestingPerson"]["edges"][0]["node"]
+        name: str = response_person["name"]["value"]
+        animal_names = sorted(
+            animal["node"]["name"]["value"] for animal in data["TestingPerson"]["edges"][0]["node"]["animals"]["edges"]
+        )
+
+        return {"person": name, "pets": animal_names}

tests/fixtures/repos/ctl_integration/transforms/converted.py

@@ -0,0 +1,18 @@
+from operator import itemgetter
+from typing import Any
+
+from infrahub_sdk.transforms import InfrahubTransform
+
+
+class ConvertedAnimalPerson(InfrahubTransform):
+    query = "animal_person"
+
+    async def transform(self, data) -> dict[str, Any]:
+        response_person = data["TestingPerson"]["edges"][0]["node"]
+        name: str = response_person["name"]["value"]
+        person = self.store.get(key=name, kind="TestingPerson")
+
+        animals = [{"type": animal.peer.typename, "name": animal.peer.name.value} for animal in person.animals.peers]
+        animals.sort(key=itemgetter("name"))
+
+        return {"person": person.name.value, "herd_size": len(animals), "animals": animals}

tests/integration/test_infrahubctl.py

@@ -0,0 +1,163 @@
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import tempfile
+from collections.abc import Generator
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import pytest
+from typer.testing import CliRunner
+
+from infrahub_sdk.ctl import config
+from infrahub_sdk.ctl.cli_commands import app, generator
+from infrahub_sdk.ctl.parameters import load_configuration
+from infrahub_sdk.repository import GitRepoManager
+from infrahub_sdk.testing.docker import TestInfrahubDockerClient
+from infrahub_sdk.testing.schemas.animal import SchemaAnimal
+from tests.helpers.utils import change_directory, strip_color
+
+if TYPE_CHECKING:
+    from infrahub_sdk import InfrahubClient
+
+FIXTURE_BASE_DIR = Path(Path(os.path.abspath(__file__)).parent / ".." / "fixtures")
+
+
+runner = CliRunner()
+
+
+class TestInfrahubCtl(TestInfrahubDockerClient, SchemaAnimal):
+    @pytest.fixture(scope="class")
+    async def base_dataset(
+        self,
+        client: InfrahubClient,
+        load_schema,
+        person_liam,
+        person_ethan,
+        person_sophia,
+        cat_luna,
+        cat_bella,
+        dog_daisy,
+        dog_rocky,
+        ctl_client_config,
+    ):
+        await client.branch.create(branch_name="branch01")
+
+    @pytest.fixture(scope="class")
+    def repository(self) -> Generator[str]:
+        temp_dir = tempfile.mkdtemp()
+
+        try:
+            fixture_path = Path(FIXTURE_BASE_DIR / "repos" / "ctl_integration")
+            shutil.copytree(fixture_path, temp_dir, dirs_exist_ok=True)
+            # Initialize fixture as git repository. This is necessary to run some infrahubctl commands.
+            GitRepoManager(temp_dir)
+
+            yield temp_dir
+
+        finally:
+            shutil.rmtree(temp_dir)
+
+    @pytest.fixture(scope="class")
+    def ctl_client_config(self, client: InfrahubClient) -> Generator:
+        load_configuration(value="infrahubctl.toml")
+        assert config.SETTINGS._settings
+        config.SETTINGS._settings.server_address = client.config.address
+        original_username = os.environ.get("INFRAHUB_USERNAME")
+        original_password = os.environ.get("INFRAHUB_PASSWORD")
+        if client.config.username and client.config.password:
+            os.environ["INFRAHUB_USERNAME"] = client.config.username
+            os.environ["INFRAHUB_PASSWORD"] = client.config.password
+        yield
+        if original_username:
+            os.environ["INFRAHUB_USERNAME"] = original_username
+        if original_password:
+            os.environ["INFRAHUB_PASSWORD"] = original_password
+
+    def test_infrahubctl_transform_cmd_animal_person(self, repository: str, base_dataset: None) -> None:
+        """Test infrahubctl transform without converting nodes."""
+
+        with change_directory(repository):
+            ethans_output = runner.invoke(app, ["transform", "animal_person", "name=Ethan Carter"])
+            structured_ethan_output = json.loads(strip_color(ethans_output.stdout))
+
+            liams_output = runner.invoke(app, ["transform", "animal_person", "name=Liam Walker"])
+            structured_liam_output = json.loads(strip_color(liams_output.stdout))
+
+            assert structured_ethan_output == {"person": "Ethan Carter", "pets": ["Bella", "Daisy", "Luna"]}
+            assert structured_liam_output == {"person": "Liam Walker", "pets": []}
+
+    def test_infrahubctl_transform_cmd_convert_animal_person(self, repository: str, base_dataset: None) -> None:
+        """Test infrahubctl transform when converting nodes."""
+
+        with change_directory(repository):
+            ethans_output = runner.invoke(app, ["transform", "animal_person_converted", "name=Ethan Carter"])
+            structured_ethan_output = json.loads(strip_color(ethans_output.stdout))
+
+            liams_output = runner.invoke(app, ["transform", "animal_person_converted", "name=Liam Walker"])
+            structured_liam_output = json.loads(strip_color(liams_output.stdout))
+
+            assert structured_ethan_output == {
+                "animals": [
+                    {"name": "Bella", "type": "TestingCat"},
+                    {"name": "Daisy", "type": "TestingDog"},
+                    {"name": "Luna", "type": "TestingCat"},
+                ],
+                "herd_size": 3,
+                "person": "Ethan Carter",
+            }
+            assert structured_liam_output == {
+                "animals": [],
+                "herd_size": 0,
+                "person": "Liam Walker",
+            }
+
+    async def test_infrahubctl_generator_cmd_animal_tags(
+        self, repository: str, base_dataset: None, client: InfrahubClient
+    ) -> None:
+        """Test infrahubctl generator without converting nodes."""
+
+        expected_generated_tags = ["raw-ethan-carter-bella", "raw-ethan-carter-daisy", "raw-ethan-carter-luna"]
+        initial_tags = await client.all(kind="BuiltinTag")
+
+        with change_directory(repository):
+            await generator(
+                generator_name="animal_tags", variables=["name=Ethan Carter"], list_available=False, path="."
+            )
+
+        final_tags = await client.all(kind="BuiltinTag")
+
+        initial_tag_names = [tag.name.value for tag in initial_tags]
+        final_tag_names = [tag.name.value for tag in final_tags]
+
+        for tag in expected_generated_tags:
+            assert tag not in initial_tag_names
+            assert tag in final_tag_names
+
+    async def test_infrahubctl_generator_cmd_animal_tags_convert_query(
+        self, repository: str, base_dataset: None, client: InfrahubClient
+    ) -> None:
+        """Test infrahubctl generator with conversion of nodes."""
+
+        expected_generated_tags = [
+            "converted-ethan-carter-bella",
+            "converted-ethan-carter-daisy",
+            "converted-ethan-carter-luna",
+        ]
+        initial_tags = await client.all(kind="BuiltinTag")
+
+        with change_directory(repository):
+            await generator(
+                generator_name="animal_tags_convert", variables=["name=Ethan Carter"], list_available=False, path="."
+            )
+
+        final_tags = await client.all(kind="BuiltinTag")
+
+        initial_tag_names = [tag.name.value for tag in initial_tags]
+        final_tag_names = [tag.name.value for tag in final_tags]
+
+        for tag in expected_generated_tags:
+            assert tag not in initial_tag_names
+            assert tag in final_tag_names