Relocate Bugzilla models to bugzilla namespace (#803)

* Ignore f401 in __init__.py files

* Reexport models from bugzilla.__init__

* Make `lookup_action` a function, rather than a bug method

* Refactor add_link_to_jira to avoid TYPE_CHECKING import

* Remove BugzillaWebhookRequest typehints from tests

* Remove redundant `Bugzilla*` names from Bugzilla models

Author: grahamalama

jbi/bugzilla/__init__.py

@@ -1,2 +1,7 @@
-from .service import BugzillaService as BugzillaService
-from .service import get_service as get_service
+from .models import (
+    Bug,
+    BugId,
+    WebhookEvent,
+    WebhookRequest,
+)
+from .service import BugzillaService, get_service

jbi/bugzilla/client.py

@@ -4,12 +4,13 @@
 
 from jbi import environment
 from jbi.common.instrument import instrument
-from jbi.models import (
-    BugzillaApiResponse,
-    BugzillaBug,
-    BugzillaComment,
+
+from .models import (
+    ApiResponse,
+    Bug,
     BugzillaComments,
-    BugzillaWebhooksResponse,
+    Comment,
+    WebhooksResponse,
 )
 
 settings = environment.get_settings()
@@ -67,12 +68,12 @@ def logged_in(self) -> bool:
         return "id" in resp
 
     @instrumented_method
-    def get_bug(self, bugid) -> BugzillaBug:
+    def get_bug(self, bugid) -> Bug:
         """Retrieve details about the specified bug id."""
         # https://bugzilla.readthedocs.io/en/latest/api/core/v1/bug.html#rest-single-bug
         url = f"{self.base_url}/rest/bug/{bugid}"
         bug_info = self._call("GET", url)
-        parsed = BugzillaApiResponse.model_validate(bug_info)
+        parsed = ApiResponse.model_validate(bug_info)
         if not parsed.bugs:
             raise BugzillaClientError(
                 f"Unexpected response content from 'GET {url}' (no 'bugs' field)"
@@ -88,7 +89,7 @@ def get_bug(self, bugid) -> BugzillaBug:
         return bug
 
     @instrumented_method
-    def get_comments(self, bugid) -> list[BugzillaComment]:
+    def get_comments(self, bugid) -> list[Comment]:
         """Retrieve the list of comments of the specified bug id."""
         # https://bugzilla.readthedocs.io/en/latest/api/core/v1/comment.html#rest-comments
         url = f"{self.base_url}/rest/bug/{bugid}/comment"
@@ -101,12 +102,12 @@ def get_comments(self, bugid) -> list[BugzillaComment]:
         return BugzillaComments.validate_python(comments)
 
     @instrumented_method
-    def update_bug(self, bugid, **fields) -> BugzillaBug:
+    def update_bug(self, bugid, **fields) -> Bug:
         """Update the specified fields of the specified bug."""
         # https://bugzilla.readthedocs.io/en/latest/api/core/v1/bug.html#rest-update-bug
         url = f"{self.base_url}/rest/bug/{bugid}"
         updated_info = self._call("PUT", url, json=fields)
-        parsed = BugzillaApiResponse.model_validate(updated_info)
+        parsed = ApiResponse.model_validate(updated_info)
         if not parsed.bugs:
             raise BugzillaClientError(
                 f"Unexpected response content from 'PUT {url}' (no 'bugs' field)"
@@ -118,7 +119,7 @@ def list_webhooks(self):
         """List the currently configured webhooks, including their status."""
         url = f"{self.base_url}/rest/webhooks/list"
         webhooks_info = self._call("GET", url)
-        parsed = BugzillaWebhooksResponse.model_validate(webhooks_info)
+        parsed = WebhooksResponse.model_validate(webhooks_info)
         if parsed.webhooks is None:
             raise BugzillaClientError(
                 f"Unexpected response content from 'GET {url}' (no 'webhooks' field)"

jbi/bugzilla/models.py

@@ -0,0 +1,178 @@
+import datetime
+import logging
+from typing import Optional, TypedDict
+from urllib.parse import ParseResult, urlparse
+
+from pydantic import BaseModel, TypeAdapter
+
+logger = logging.getLogger(__name__)
+JIRA_HOSTNAMES = ("jira", "atlassian")
+
+BugId = TypedDict("BugId", {"id": Optional[int]})
+
+
+class WebhookUser(BaseModel, frozen=True):
+    """Bugzilla User Object"""
+
+    id: int
+    login: str
+    real_name: str
+
+
+class WebhookEventChange(BaseModel, frozen=True, coerce_numbers_to_str=True):
+    """Bugzilla Change Object"""
+
+    field: str
+    removed: str
+    added: str
+
+
+class WebhookEvent(BaseModel, frozen=True):
+    """Bugzilla Event Object"""
+
+    action: str
+    time: Optional[datetime.datetime] = None
+    user: Optional[WebhookUser] = None
+    changes: Optional[list[WebhookEventChange]] = None
+    target: Optional[str] = None
+    routing_key: Optional[str] = None
+
+    def changed_fields(self) -> list[str]:
+        """Returns the names of changed fields in a bug"""
+
+        return [c.field for c in self.changes] if self.changes else []
+
+
+class WebhookComment(BaseModel, frozen=True):
+    """Bugzilla Comment Object"""
+
+    body: Optional[str] = None
+    id: Optional[int] = None
+    number: Optional[int] = None
+    is_private: Optional[bool] = None
+    creation_time: Optional[datetime.datetime] = None
+
+
+class Bug(BaseModel, frozen=True):
+    """Bugzilla Bug Object"""
+
+    id: int
+    is_private: Optional[bool] = None
+    type: Optional[str] = None
+    product: Optional[str] = None
+    component: Optional[str] = None
+    whiteboard: Optional[str] = None
+    keywords: Optional[list] = None
+    flags: Optional[list] = None
+    groups: Optional[list] = None
+    status: Optional[str] = None
+    resolution: Optional[str] = None
+    see_also: Optional[list] = None
+    summary: Optional[str] = None
+    severity: Optional[str] = None
+    priority: Optional[str] = None
+    creator: Optional[str] = None
+    assigned_to: Optional[str] = None
+    comment: Optional[WebhookComment] = None
+
+    @property
+    def product_component(self) -> str:
+        """Return the component prefixed with the product
+        as show in the Bugzilla UI (eg. ``Core::General``).
+        """
+        result = self.product + "::" if self.product else ""
+        return result + self.component if self.component else result
+
+    def is_assigned(self) -> bool:
+        """Return `true` if the bug is assigned to a user."""
+        return self.assigned_to != "[email protected]"
+
+    def extract_from_see_also(self, project_key):
+        """Extract Jira Issue Key from see_also if jira url present"""
+        if not self.see_also or len(self.see_also) == 0:
+            return None
+
+        candidates = []
+        for url in self.see_also:
+            try:
+                parsed_url: ParseResult = urlparse(url=url)
+                host_parts = parsed_url.hostname.split(".")
+            except (ValueError, AttributeError):
+                logger.debug(
+                    "Bug %s `see_also` is not a URL: %s",
+                    self.id,
+                    url,
+                    extra={
+                        "bug": {
+                            "id": self.id,
+                        }
+                    },
+                )
+                continue
+
+            if any(part in JIRA_HOSTNAMES for part in host_parts):
+                parsed_jira_key = parsed_url.path.rstrip("/").split("/")[-1]
+                if parsed_jira_key:  # URL ending with /
+                    # Issue keys are like `{project_key}-{number}`
+                    if parsed_jira_key.startswith(f"{project_key}-"):
+                        return parsed_jira_key
+                    # If not obvious, then keep this link as candidate.
+                    candidates.append(parsed_jira_key)
+
+        return candidates[0] if candidates else None
+
+
+class WebhookRequest(BaseModel, frozen=True):
+    """Bugzilla Webhook Request Object"""
+
+    webhook_id: int
+    webhook_name: str
+    event: WebhookEvent
+    bug: Bug
+
+
+class Comment(BaseModel, frozen=True):
+    """Bugzilla Comment"""
+
+    id: int
+    text: str
+    is_private: bool
+    creator: str
+
+
+BugzillaComments = TypeAdapter(list[Comment])
+
+
+class ApiResponse(BaseModel, frozen=True):
+    """Bugzilla Response Object"""
+
+    faults: Optional[list] = None
+    bugs: Optional[list[Bug]] = None
+
+
+class Webhook(BaseModel, frozen=True):
+    """Bugzilla Webhook"""
+
+    id: int
+    name: str
+    url: str
+    event: str
+    product: str
+    component: str
+    enabled: bool
+    errors: int
+    # Ignored fields:
+    # creator: str
+
+    @property
+    def slug(self):
+        """Return readable identifier"""
+        name = self.name.replace(" ", "-").lower()
+        product = self.product.replace(" ", "-").lower()
+        return f"{self.id}-{name}-{product}"
+
+
+class WebhooksResponse(BaseModel, frozen=True):
+    """Bugzilla Webhooks List Response Object"""
+
+    webhooks: Optional[list[Webhook]] = None

jbi/bugzilla/service.py

@@ -4,14 +4,11 @@
 import requests
 from statsd.defaults.env import statsd
 
-from jbi import Operation, environment
+from jbi import environment
 from jbi.common.instrument import ServiceHealth
-from jbi.models import (
-    ActionContext,
-    BugzillaBug,
-)
 
 from .client import BugzillaClient, BugzillaClientError
+from .models import Bug
 
 settings = environment.get_settings()
 
@@ -67,18 +64,10 @@ def _all_webhooks_enabled(self):
                 return False
         return True
 
-    def add_link_to_jira(self, context: ActionContext):
-        """Add link to Jira in Bugzilla ticket"""
-        bug = context.bug
-        issue_key = context.jira.issue
-        jira_url = f"{settings.jira_base_url}browse/{issue_key}"
-        logger.debug(
-            "Link %r on Bug %s",
-            jira_url,
-            bug.id,
-            extra=context.update(operation=Operation.LINK).model_dump(),
-        )
-        return self.client.update_bug(bug.id, see_also={"add": [jira_url]})
+    def add_link_to_see_also(self, bug: Bug, link: str):
+        """Add link to Bugzilla ticket"""
+
+        return self.client.update_bug(bug.id, see_also={"add": [link]})
 
     def get_description(self, bug_id: int):
         """Fetch a bug's description
@@ -90,7 +79,7 @@ def get_description(self, bug_id: int):
         comment_body = comment_list[0].text if comment_list else ""
         return str(comment_body)
 
-    def refresh_bug_data(self, bug: BugzillaBug):
+    def refresh_bug_data(self, bug: Bug):
         """Re-fetch a bug to ensure we have the most up-to-date data"""
 
         refreshed_bug_data = self.client.get_bug(bug.id)

jbi/jira/service.py

@@ -13,9 +13,9 @@
 import requests
 from requests import exceptions as requests_exceptions
 
-from jbi import Operation, environment
+from jbi import Operation, bugzilla, environment
 from jbi.common.instrument import ServiceHealth
-from jbi.models import ActionContext, BugzillaBug
+from jbi.models import ActionContext
 
 from .client import JiraClient, JiraCreateError
 
@@ -306,7 +306,7 @@ def add_jira_comments_for_changes(self, context: ActionContext):
         return jira_response_comments
 
     def delete_jira_issue_if_duplicate(
-        self, context: ActionContext, latest_bug: BugzillaBug
+        self, context: ActionContext, latest_bug: bugzilla.Bug
     ):
         """Rollback the Jira issue creation if there is already a linked Jira issue
         on the Bugzilla ticket"""