Add event to log several revoked approvals on a PC

Author: gmazoyer

backend/infrahub/core/constants/__init__.py

@@ -66,6 +66,7 @@ class EventType(InfrahubStringEnum):
     PROPOSED_CHANGE_APPROVED = f"{EVENT_NAMESPACE}.proposed_change.approved"
     PROPOSED_CHANGE_REJECTED = f"{EVENT_NAMESPACE}.proposed_change.rejected"
     PROPOSED_CHANGE_APPROVAL_REVOKED = f"{EVENT_NAMESPACE}.proposed_change.approval_revoked"
+    PROPOSED_CHANGE_APPROVALS_REVOKED = f"{EVENT_NAMESPACE}.proposed_change.approvals_revoked"
     PROPOSED_CHANGE_REJECTION_REVOKED = f"{EVENT_NAMESPACE}.proposed_change.rejection_revoked"
     PROPOSED_CHANGE_THREAD_CREATED = f"{EVENT_NAMESPACE}.proposed_change_thread.created"
     PROPOSED_CHANGE_THREAD_UPDATED = f"{EVENT_NAMESPACE}.proposed_change_thread.updated"

backend/infrahub/events/__init__.py

@@ -5,6 +5,7 @@
 from .node_action import NodeCreatedEvent, NodeDeletedEvent, NodeUpdatedEvent
 from .proposed_change_action import (
     ProposedChangeApprovalRevokedEvent,
+    ProposedChangeApprovalsRevokedEvent,
     ProposedChangeApprovedEvent,
     ProposedChangeMergedEvent,
     ProposedChangeRejectedEvent,
@@ -32,6 +33,7 @@
     "NodeDeletedEvent",
     "NodeUpdatedEvent",
     "ProposedChangeApprovalRevokedEvent",
+    "ProposedChangeApprovalsRevokedEvent",
     "ProposedChangeApprovedEvent",
     "ProposedChangeMergedEvent",
     "ProposedChangeRejectedEvent",

backend/infrahub/events/proposed_change_action.py

@@ -1,6 +1,6 @@
-from typing import ClassVar
+from typing import ClassVar, Self
 
-from pydantic import Field
+from pydantic import Field, model_validator
 
 from infrahub.core.constants import InfrahubKind, MutationAction
 
@@ -150,6 +150,33 @@ class ProposedChangeRejectionRevokedEvent(ProposedChangeReviewRevokedEvent):
     event_name: ClassVar[str] = f"{EVENT_NAMESPACE}.proposed_change.rejection_revoked"
 
 
+class ProposedChangeApprovalsRevokedEvent(ProposedChangeEvent):
+    reviewer_account_ids: list[str] = Field(..., description="IDs of accounts whose approval was revoked")
+    reviewer_account_names: list[str] = Field(..., description="Names of accounts whose approval was revoked")
+
+    event_name: ClassVar[str] = f"{EVENT_NAMESPACE}.proposed_change.approvals_revoked"
+
+    @model_validator(mode="after")
+    def check_same_length(self) -> Self:
+        if len(self.reviewer_account_ids) != len(self.reviewer_account_names):
+            raise ValueError("reviewer_account_ids and reviewer_account_names must have the same number of items")
+        return self
+
+    def get_related(self) -> list[dict[str, str]]:
+        related = super().get_related()
+        for i, account_id in enumerate(self.reviewer_account_ids):
+            related.append(
+                {
+                    "prefect.resource.id": account_id,
+                    "prefect.resource.role": "infrahub.related.node",
+                    "infrahub.node.kind": InfrahubKind.GENERICACCOUNT,
+                    "infrahub.node.id": account_id,
+                    "infrahub.reviewer.account.name": self.reviewer_account_names[i],
+                }
+            )
+        return related
+
+
 class ProposedChangeThreadEvent(ProposedChangeEvent):
     thread_id: str = Field(..., description="The ID of the thread that was created or updated")
     thread_kind: str = Field(..., description="The name of the thread that was created or updated")