Add more functional tests

Author: dgarros

backend/infrahub/core/constants/__init__.py

@@ -63,6 +63,9 @@ class EventType(InfrahubStringEnum):
 
     REPOSITORY_UPDATE_COMMIT = f"{EVENT_NAMESPACE}.repository.update_commit"
 
+    ARTIFACT_CREATED = f"{EVENT_NAMESPACE}.artifact.created"
+    ARTIFACT_UPDATED = f"{EVENT_NAMESPACE}.artifact.updated"
+
 
 class PermissionLevel(enum.Flag):
     READ = 1

backend/infrahub/git/integrator.py

@@ -1150,7 +1150,7 @@ async def execute_python_check(
 
     @task(name="python-transform-execute", task_run_name="Execute Python Transform", cache_policy=NONE)  # type: ignore[arg-type]
     async def execute_python_transform(
-        self, branch_name: str, commit: str, location: str, client: InfrahubClient, data: Optional[dict] = None
+        self, branch_name: str, commit: str, location: str, client: InfrahubClient, data: dict | None = None
     ) -> Any:
         """Execute A Python Transform stored in the repository."""
         log = get_run_logger()

backend/infrahub/message_bus/operations/event/worker.py

@@ -7,5 +7,3 @@
 @flow(name="event-worker-newprimary-api")
 async def new_primary_api(message: messages.EventWorkerNewPrimaryAPI, service: InfrahubServices) -> None:
     service.log.info("api_worker promoted to primary", worker_id=message.worker_id)
-
-    # NOTE Should we remove this message ?

backend/infrahub/webhook/models.py

@@ -6,20 +6,19 @@
 from typing import TYPE_CHECKING, Any, Optional, Union
 from uuid import UUID, uuid4
 
-from infrahub_sdk.protocols import CoreCustomWebhook, CoreStandardWebhook, CoreTransformPython, CoreWebhook
 from pydantic import BaseModel, ConfigDict, Field, computed_field
 from typing_extensions import Self
 
 from infrahub.core import registry
 from infrahub.core.constants import InfrahubKind
 from infrahub.core.timestamp import Timestamp
 from infrahub.git.repository import InfrahubReadOnlyRepository, InfrahubRepository
-from infrahub.transformations.constants import DEFAULT_TRANSFORM_TIMEOUT
 from infrahub.trigger.models import EventTrigger, ExecuteWorkflow, TriggerDefinition, TriggerType
 from infrahub.workflows.catalogue import WEBHOOK_PROCESS
 
 if TYPE_CHECKING:
     from httpx import Response
+    from infrahub_sdk.protocols import CoreCustomWebhook, CoreStandardWebhook, CoreTransformPython, CoreWebhook
 
     from infrahub.services import InfrahubServices
 
@@ -79,15 +78,17 @@ class EventContext(BaseModel):
 
     @classmethod
     def from_event(cls, event_id: str, event_type: str, event_occured_at: str, event_payload: dict[str, Any]) -> Self:
-        """Extract the context from the raw  eventwe are getting from Prefect."""
+        """Extract the context from the raw event we are getting from Prefect."""
 
         infrahub_context: dict[str, Any] = event_payload.get("context", {})
         account_info: dict[str, Any] = infrahub_context.get("account", {})
         branch_info: dict[str, Any] = infrahub_context.get("branch", {})
 
         return cls(
             id=event_id,
-            branch=branch_info.get("name") if branch_info else None,
+            branch=branch_info.get("name")
+            if branch_info and branch_info.get("name") != registry.get_global_branch().name
+            else None,
             account_id=account_info.get("account_id"),
             occured_at=event_occured_at,
             event=event_type,
@@ -114,10 +115,16 @@ def _assign_headers(self) -> None:
     def webhook_type(self) -> str:
         return self.__class__.__name__
 
-    async def send(self, data: dict[str, Any], context: EventContext, service: InfrahubServices) -> Response:
+    async def prepare(self, data: dict[str, Any], context: EventContext, service: InfrahubServices) -> None:
         await self._prepare_payload(data=data, context=context, service=service)
         self._assign_headers()
-        return await service.http.post(url=self.url, json=self._payload, headers=self._headers)
+
+    async def send(self, data: dict[str, Any], context: EventContext, service: InfrahubServices) -> Response:
+        await self.prepare(data=data, context=context, service=service)
+        return await service.http.post(url=self.url, json=self.get_payload(), headers=self._headers)
+
+    def get_payload(self) -> dict[str, Any]:
+        return self._payload
 
     def to_cache(self) -> dict[str, Any]:
         return self.model_dump()
@@ -179,6 +186,7 @@ class TransformWebhook(Webhook):
     transform_name: str = Field(...)
     transform_class: str = Field(...)
     transform_file: str = Field(...)
+    transform_timeout: int = Field(...)
 
     async def _prepare_payload(self, data: dict[str, Any], context: EventContext, service: InfrahubServices) -> None:
         repo: Union[InfrahubReadOnlyRepository, InfrahubRepository]
@@ -197,17 +205,11 @@ async def _prepare_payload(self, data: dict[str, Any], context: EventContext, se
                 service=service,
             )
 
-        default_branch = repo.default_branch
-        commit = repo.get_commit_value(branch_name=default_branch)
-
-        timeout = DEFAULT_TRANSFORM_TIMEOUT
-        if transform := await service.client.get(
-            kind=CoreTransformPython, name__value=self.transform_name, raise_when_missing=False
-        ):
-            timeout = transform.timeout.value
+        branch = context.branch or repo.default_branch
+        commit = repo.get_commit_value(branch_name=branch)
 
-        self._payload = await repo.execute_python_transform.with_options(timeout_seconds=timeout)(
-            branch_name=default_branch,
+        self._payload = await repo.execute_python_transform.with_options(timeout_seconds=self.transform_timeout)(
+            branch_name=branch,
             commit=commit,
             location=f"{self.transform_file}::{self.transform_class}",
             data={"data": data, **context.model_dump()},
@@ -227,4 +229,5 @@ def from_object(cls, obj: CoreCustomWebhook, transform: CoreTransformPython) ->
             transform_name=transform.name.value,
             transform_class=transform.class_name.value,
             transform_file=transform.file_path.value,
+            transform_timeout=transform.timeout.value,
         )

backend/infrahub/webhook/tasks.py

@@ -120,13 +120,13 @@ async def configure_webhook_all(service: InfrahubServices) -> None:
 
 
 @flow(name="webhook-setup-automation-one", flow_run_name="Configuration webhook automation for one webhook")
-async def configure_webhook_one(event_data: dict, service: InfrahubServices) -> None:
+async def configure_webhook_one(event_type: str, event_data: dict, service: InfrahubServices) -> None:
     log = get_run_logger()
 
     webhook = await service.client.get(kind=CoreWebhook, id=event_data["node_id"])
     trigger = WebhookTriggerDefinition.from_object(webhook)
 
-    # TODO: Need to check in the event if the node was created, updated or deleted
+    delete_automation: bool = "infrahub.node.deleted" in event_type
 
     async with get_client(sync_client=False) as prefect_client:
         # Query the deployment associated with the trigger to have its ID
@@ -143,10 +143,14 @@ async def configure_webhook_one(event_data: dict, service: InfrahubServices) ->
 
         existing_automations = await prefect_client.read_automations_by_name(trigger.generate_name())
 
-        if existing_automations:
-            automation = existing_automations[0]
-            await prefect_client.update_automation(automation_id=automation.id, automation=automation)
+        if existing_automations and not delete_automation:
+            existing_automation = existing_automations[0]
+            await prefect_client.update_automation(automation_id=existing_automation.id, automation=automation)
             log.info(f"Automation {trigger.generate_name()} updated")
+        elif existing_automations and delete_automation:
+            existing_automation = existing_automations[0]
+            await prefect_client.delete_automation(automation_id=existing_automation.id)
+            log.info(f"Automation {trigger.generate_name()} deleted")
         else:
             await prefect_client.create_automation(automation=automation)
             log.info(f"Automation {trigger.generate_name()} created")

backend/infrahub/webhook/triggers.py

@@ -16,10 +16,11 @@
         ExecuteWorkflow(
             workflow=WEBHOOK_CONFIGURE_ONE,
             parameters={
+                "event_type": "{{ event.event }}",
                 "event_data": {
                     "__prefect_kind": "json",
                     "value": {"__prefect_kind": "jinja", "template": "{{ event.payload['data'] | tojson }}"},
-                }
+                },
             },
         ),
     ],

backend/tests/fixtures/repos/car-dealership/initial__main/.infrahub.yml

@@ -29,6 +29,9 @@ python_transforms:
   - name: CarSpecMarkdown
     class_name: CarSpecMarkdown
     file_path: "transforms/car_spec_markdown.py"
+  - name: WebhookTransformer
+    class_name: WebhookTransformer
+    file_path: "transforms/webhook_transformer.py"
 
 artifact_definitions:
   - name: "Ownership report"

backend/tests/fixtures/repos/car-dealership/initial__main/transforms/webhook_transformer.py

@@ -0,0 +1,11 @@
+from typing import Any
+
+from infrahub_sdk.transforms import InfrahubTransform
+
+
+class WebhookTransformer(InfrahubTransform):
+    query = "person_with_cars"
+    timeout = 5
+
+    async def transform(self, data: dict[str, Any]) -> dict[str, Any]:
+        return {key.upper(): value for key, value in data.items()}