Add testing directory based on infrahub_testcontainers

infrahub_sdk/schema/main.py

@@ -4,7 +4,7 @@
 from enum import Enum
 from typing import TYPE_CHECKING, Any, Optional, Union
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 if TYPE_CHECKING:
     from ..node import InfrahubNode, InfrahubNodeSync
@@ -89,6 +89,8 @@
 
 
 class AttributeSchema(BaseModel):
+    model_config = ConfigDict(use_enum_values=True)
+
     id: Optional[str] = None
     state: SchemaState = SchemaState.PRESENT
     name: str
@@ -108,12 +110,16 @@
 
 
 class AttributeSchemaAPI(AttributeSchema):
+    model_config = ConfigDict(use_enum_values=True)
+
     inherited: bool = False
     read_only: bool = False
     allow_override: AllowOverrideType = AllowOverrideType.ANY
 
 
 class RelationshipSchema(BaseModel):
+    model_config = ConfigDict(use_enum_values=True)
+
     id: Optional[str] = None
     state: SchemaState = SchemaState.PRESENT
     name: str
@@ -133,6 +139,8 @@
 
 
 class RelationshipSchemaAPI(RelationshipSchema):
+    model_config = ConfigDict(use_enum_values=True)
+
     inherited: bool = False
     read_only: bool = False
     hierarchical: Optional[str] = None
@@ -245,6 +253,8 @@
 
 
 class BaseSchema(BaseModel):
+    model_config = ConfigDict(use_enum_values=True)
+
     id: Optional[str] = None
     state: SchemaState = SchemaState.PRESENT
     name: str
@@ -275,6 +285,8 @@
 
 
 class BaseNodeSchema(BaseSchema):
+    model_config = ConfigDict(use_enum_values=True)
+
     inherit_from: list[str] = Field(default_factory=list)
     branch: Optional[BranchSupportType] = None
     default_filter: Optional[str] = None
@@ -299,6 +311,8 @@
 
 
 class NodeExtensionSchema(BaseModel):
+    model_config = ConfigDict(use_enum_values=True)
+
     name: Optional[str] = None
     kind: str
     description: Optional[str] = None
@@ -311,13 +325,20 @@
 
 
 class SchemaRoot(BaseModel):
+    model_config = ConfigDict(use_enum_values=True)
+
     version: str
     generics: list[GenericSchema] = Field(default_factory=list)
     nodes: list[NodeSchema] = Field(default_factory=list)
     node_extensions: list[NodeExtensionSchema] = Field(default_factory=list)
 
+    def to_schema_dict(self) -> dict[str, Any]:
+        return self.model_dump(exclude_unset=True, exclude_defaults=True)
+
 
 class SchemaRootAPI(BaseModel):
+    model_config = ConfigDict(use_enum_values=True)
+
     version: str
     generics: list[GenericSchemaAPI] = Field(default_factory=list)
     nodes: list[NodeSchemaAPI] = Field(default_factory=list)

infrahub_sdk/testing/docker.py

@@ -0,0 +1,18 @@
+import pytest
+from infrahub_testcontainers.helpers import TestInfrahubDocker
+
+from .. import Config, InfrahubClient, InfrahubClientSync
+
+
+class TestInfrahubDockerClient(TestInfrahubDocker):
+    @pytest.fixture(scope="class")
+    def client(self, infrahub_port: int) -> InfrahubClient:
+        return InfrahubClient(
+            config=Config(username="admin", password="infrahub", address=f"http://localhost:{infrahub_port}")  # noqa: S106
+        )
+
+    @pytest.fixture(scope="class")
+    def client_sync(self, infrahub_port: int) -> InfrahubClientSync:
+        return InfrahubClientSync(
+            config=Config(username="admin", password="infrahub", address=f"http://localhost:{infrahub_port}")  # noqa: S106
+        )

infrahub_sdk/testing/repository.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: Optional[str] = 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: Optional[str] = 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/schemas/car_person.py

@@ -0,0 +1,145 @@
+import pytest
+
+from infrahub_sdk import InfrahubClient
+from infrahub_sdk.node import InfrahubNode
+from infrahub_sdk.schema.main import AttributeKind, NodeSchema, RelationshipKind, SchemaRoot
+from infrahub_sdk.schema.main import AttributeSchema as Attr
+from infrahub_sdk.schema.main import RelationshipSchema as Rel
+
+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) -> NodeSchema:
+        return NodeSchema(
+            name="Person",
+            namespace=NAMESPACE,
+            include_in_menu=True,
+            label="Person",
+            human_friendly_id=["name__value"],
+            attributes=[
+                Attr(name="name", kind=AttributeKind.TEXT, unique=True),
+                Attr(name="description", kind=AttributeKind.TEXT, optional=True),
+                Attr(name="height", kind=AttributeKind.NUMBER, optional=True),
+                Attr(name="age", kind=AttributeKind.NUMBER, optional=True),
+            ],
+            relationships=[
+                Rel(name="cars", kind=RelationshipKind.GENERIC, optional=True, peer=TESTING_CAR, cardinality="many")
+            ],
+        )
+
+    @pytest.fixture(scope="class")
+    def schema_car_base(self) -> NodeSchema:
+        return NodeSchema(
+            name="Car",
+            namespace=NAMESPACE,
+            include_in_menu=True,
+            default_filter="name__value",
+            human_friendly_id=["owner__name__value", "name__value"],
+            label="Car",
+            attributes=[
+                Attr(name="name", kind=AttributeKind.TEXT),
+                Attr(name="description", kind=AttributeKind.TEXT, optional=True),
+                Attr(name="color", kind=AttributeKind.TEXT),
+            ],
+            relationships=[
+                Rel(
+                    name="owner",
+                    kind=RelationshipKind.ATTRIBUTE,
+                    optional=False,
+                    peer=TESTING_PERSON,
+                    cardinality="one",
+                ),
+                Rel(
+                    name="manufacturer",
+                    kind=RelationshipKind.ATTRIBUTE,
+                    optional=False,
+                    peer=TESTING_MANUFACTURER,
+                    cardinality="one",
+                    identifier="car__manufacturer",
+                ),
+            ],
+        )
+
+    @pytest.fixture(scope="class")
+    def schema_manufacturer_base(self) -> NodeSchema:
+        return NodeSchema(
+            name="Manufacturer",
+            namespace=NAMESPACE,
+            include_in_menu=True,
+            label="Manufacturer",
+            human_friendly_id=["name__value"],
+            attributes=[
+                Attr(name="name", kind=AttributeKind.TEXT),
+                Attr(name="description", kind=AttributeKind.TEXT, optional=True),
+            ],
+            relationships=[
+                Rel(
+                    name="cars",
+                    kind=RelationshipKind.GENERIC,
+                    optional=True,
+                    peer=TESTING_CAR,
+                    cardinality="many",
+                    identifier="car__manufacturer",
+                ),
+                Rel(
+                    name="customers",
+                    kind=RelationshipKind.GENERIC,
+                    optional=True,
+                    peer=TESTING_PERSON,
+                    cardinality="many",
+                    identifier="person__manufacturer",
+                ),
+            ],
+        )
+
+    @pytest.fixture(scope="class")
+    def schema_base(
+        self,
+        schema_car_base: NodeSchema,
+        schema_person_base: NodeSchema,
+        schema_manufacturer_base: NodeSchema,
+    ) -> SchemaRoot:
+        return SchemaRoot(version="1.0", nodes=[schema_car_base, schema_person_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}