Add testing directory

Author: dgarros

infrahub_sdk/testing/__init__.py

@@ -0,0 +1,14 @@
+import pytest
+from infrahub_testing.helpers import TestInfrahubDocker
+
+from .. import Config, InfrahubClient, InfrahubClientSync
+
+
+class TestInfrahubDockerClient(TestInfrahubDocker):
+    @pytest.fixture(scope="class")
+    def infrahub_client(self, infrahub_port: int) -> InfrahubClient:
+        return InfrahubClient(config=Config(address=f"http://localhost:{infrahub_port}"))
+
+    @pytest.fixture(scope="class")
+    def infrahub_client_sync(self, infrahub_port: int) -> InfrahubClientSync:
+        return InfrahubClientSync(config=Config(address=f"http://localhost:{infrahub_port}"))

infrahub_sdk/testing/docker.py

@@ -0,0 +1,103 @@
+import asyncio
+import shutil
+from dataclasses import dataclass, field
+from enum import Enum
+from pathlib import Path
+from typing import Optional
+
+from git.repo import Repo
+
+from infrahub_sdk import InfrahubClient
+from infrahub_sdk.graphql import Mutation
+from infrahub_sdk.protocols import CoreGenericRepository
+
+
+# NOTE we shouldn't duplicate this, need to figure out a better solution
+class RepositorySyncStatus(str, Enum):
+    UNKNOWN = "unknown"
+    IN_SYNC = "in-sync"
+    ERROR_IMPORT = "error-import"
+    SYNCING = "syncing"
+
+
+class GitRepoType(str, Enum):
+    INTEGRATED = "CoreRepository"
+    READ_ONLY = "CoreReadOnlyRepository"
+
+
+@dataclass
+class GitRepo:
+    name: str
+    src_directory: Path
+    dst_directory: Path
+
+    type: GitRepoType = GitRepoType.INTEGRATED
+
+    _repo: Optional[Repo] = None
+    initial_branch: str = "main"
+    directories_to_ignore: list[str] = field(default_factory=list)
+    remote_directory_name: str = "/remote"
+    _branches: list[str] = field(default_factory=list)
+
+    @property
+    def repo(self) -> Repo:
+        if self._repo:
+            return self._repo
+        raise ValueError("Repo hasn't been initialized yet")
+
+    def __post_init__(self) -> None:
+        self.init()
+
+    @property
+    def path(self) -> str:
+        return str(self.src_directory / self.name)
+
+    def init(self) -> None:
+        shutil.copytree(
+            src=self.src_directory,
+            dst=self.dst_directory / self.name,
+            ignore=shutil.ignore_patterns(".git"),
+        )
+        self._repo = Repo.init(self.dst_directory / self.name, initial_branch=self.initial_branch)
+        for untracked in self.repo.untracked_files:
+            self.repo.index.add(untracked)
+        self.repo.index.commit("First commit")
+
+        self.repo.git.checkout(self.initial_branch)
+
+    async def add_to_infrahub(self, client: InfrahubClient, branch: str | None = None) -> dict:
+        input_data = {
+            "data": {
+                "name": {"value": self.name},
+                "location": {"value": f"{self.remote_directory_name}/{self.name}"},
+            },
+        }
+
+        query = Mutation(
+            mutation=f"{self.type.value}Create",
+            input_data=input_data,
+            query={"ok": None},
+        )
+
+        return await client.execute_graphql(
+            query=query.render(), branch_name=branch or self.initial_branch, tracker="mutation-repository-create"
+        )
+
+    async def wait_for_sync_to_complete(
+        self, client: InfrahubClient, branch: str | None = None, interval: int = 5, retries: int = 6
+    ) -> bool:
+        for _ in range(retries):
+            repo = await client.get(
+                kind=CoreGenericRepository,  # type: ignore[type-abstract]
+                name__value=self.name,
+                branch=branch or self.initial_branch,
+            )
+            status = repo.sync_status.value
+            if status == RepositorySyncStatus.IN_SYNC.value:
+                return True
+            if status == RepositorySyncStatus.ERROR_IMPORT.value:
+                return False
+
+            await asyncio.sleep(interval)
+
+        return False

infrahub_sdk/testing/repository.py

@@ -0,0 +1,144 @@
+from typing import Any
+
+import pytest
+
+from infrahub_sdk import InfrahubClient
+from infrahub_sdk.node import InfrahubNode
+
+NAMESPACE = "Testing"
+
+TESTING_MANUFACTURER = f"{NAMESPACE}Manufacturer"
+TESTING_PERSON = f"{NAMESPACE}Person"
+TESTING_CAR = f"{NAMESPACE}Car"
+
+
+class SchemaCarPerson:
+    @pytest.fixture(scope="class")
+    def schema_person_base(self) -> dict[str, Any]:
+        return {
+            "name": "Person",
+            "namespace": NAMESPACE,
+            "include_in_menu": True,
+            "label": "Person",
+            "human_friendly_id": ["name__value"],
+            "attributes": [
+                {"name": "name", "kind": "Text", "unique": True},
+                {"name": "description", "kind": "Text", "optional": True},
+                {"name": "height", "kind": "Number", "optional": True},
+                {"name": "age", "kind": "Number", "optional": True},
+            ],
+            "relationships": [
+                {"name": "cars", "kind": "Generic", "optional": True, "peer": TESTING_CAR, "cardinality": "many"}
+            ],
+        }
+
+    @pytest.fixture(scope="class")
+    def schema_car_base(self) -> dict[str, Any]:
+        return {
+            "name": "Car",
+            "namespace": NAMESPACE,
+            "include_in_menu": True,
+            "default_filter": "name__value",
+            "human_friendly_id": ["owner__name__value", "name__value"],
+            "label": "Car",
+            "attributes": [
+                {"name": "name", "kind": "Text"},
+                {"name": "description", "kind": "Text", "optional": True},
+                {"name": "color", "kind": "Text"},
+            ],
+            "relationships": [
+                {
+                    "name": "owner",
+                    "kind": "Attribute",
+                    "optional": False,
+                    "peer": TESTING_PERSON,
+                    "cardinality": "one",
+                },
+                {
+                    "name": "manufacturer",
+                    "kind": "Attribute",
+                    "optional": False,
+                    "peer": TESTING_MANUFACTURER,
+                    "cardinality": "one",
+                    "identifier": "car__manufacturer",
+                },
+            ],
+        }
+
+    @pytest.fixture(scope="class")
+    def schema_manufacturer_base(self) -> dict[str, Any]:
+        return {
+            "name": "Manufacturer",
+            "namespace": NAMESPACE,
+            "include_in_menu": True,
+            "label": "Manufacturer",
+            "human_friendly_id": ["name__value"],
+            "attributes": [{"name": "name", "kind": "Text"}, {"name": "description", "kind": "Text", "optional": True}],
+            "relationships": [
+                {
+                    "name": "cars",
+                    "kind": "Generic",
+                    "optional": True,
+                    "peer": TESTING_CAR,
+                    "cardinality": "many",
+                    "identifier": "car__manufacturer",
+                },
+                {
+                    "name": "customers",
+                    "kind": "Generic",
+                    "optional": True,
+                    "peer": TESTING_PERSON,
+                    "cardinality": "many",
+                    "identifier": "person__manufacturer",
+                },
+            ],
+        }
+
+    @pytest.fixture(scope="class")
+    def schema_base(
+        self,
+        schema_car_base: dict[str, Any],
+        schema_person_base: dict[str, Any],
+        schema_manufacturer_base: dict[str, Any],
+    ) -> dict[str, Any]:
+        return {
+            "version": "1.0",
+            "nodes": [schema_person_base, schema_car_base, schema_manufacturer_base],
+        }
+
+    async def create_persons(self, client: InfrahubClient, branch: str) -> list[InfrahubNode]:
+        john = await client.create(kind=TESTING_PERSON, name="John Doe", branch=branch)
+        await john.save()
+
+        jane = await client.create(kind=TESTING_PERSON, name="Jane Doe", branch=branch)
+        await jane.save()
+
+        return [john, jane]
+
+    async def create_manufacturers(self, client: InfrahubClient, branch: str) -> list[InfrahubNode]:
+        obj1 = await client.create(kind=TESTING_MANUFACTURER, name="Volkswagen", branch=branch)
+        await obj1.save()
+
+        obj2 = await client.create(kind=TESTING_MANUFACTURER, name="Renault", branch=branch)
+        await obj2.save()
+
+        obj3 = await client.create(kind=TESTING_MANUFACTURER, name="Mercedes", branch=branch)
+        await obj3.save()
+
+        return [obj1, obj2, obj3]
+
+    async def create_initial_data(self, client: InfrahubClient, branch: str) -> dict[str, list[InfrahubNode]]:
+        persons = await self.create_persons(client=client, branch=branch)
+        manufacturers = await self.create_manufacturers(client=client, branch=branch)
+
+        car10 = await client.create(
+            kind=TESTING_CAR, name="Golf", color="Black", manufacturer=manufacturers[0].id, owner=persons[0].id
+        )
+        await car10.save()
+
+        car20 = await client.create(
+            kind=TESTING_CAR, name="Megane", color="Red", manufacturer=manufacturers[1].id, owner=persons[1].id
+        )
+        await car20.save()
+
+        return {TESTING_PERSON: persons, TESTING_CAR: [car10, car20], TESTING_MANUFACTURER: manufacturers}