feat(perforce): Implement repository/depot and code mapping logic

pyproject.toml

@@ -445,6 +445,7 @@ module = [
     "sentry.integrations.msteams.handlers.*",
     "sentry.integrations.opsgenie.handlers.*",
     "sentry.integrations.pagerduty.*",
+    "sentry.integrations.perforce.*",
     "sentry.integrations.project_management.*",
     "sentry.integrations.repository.*",
     "sentry.integrations.services.*",

src/sentry/integrations/perforce/client.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Sequence
+from collections.abc import Generator, Sequence
 from contextlib import contextmanager
-from typing import Any
+from typing import Any, TypedDict
 
 from P4 import P4, P4Exception
 
@@ -22,6 +22,69 @@
 
 logger = logging.getLogger(__name__)
 
+# Default buffer size when fetching changelist ranges to ensure complete coverage
+DEFAULT_REVISION_RANGE = 10
+
+
+class P4ChangeInfo(TypedDict):
+    """Type definition for Perforce changelist information."""
+
+    change: str
+    user: str
+    client: str
+    time: str
+    desc: str
+
+
+class P4DepotInfo(TypedDict):
+    """Type definition for Perforce depot information."""
+
+    name: str
+    type: str
+    description: str
+
+
+class P4UserInfo(TypedDict, total=False):
+    """Type definition for Perforce user information."""
+
+    email: str
+    full_name: str
+    username: str
+
+
+class P4CommitInfo(TypedDict):
+    """Type definition for Sentry commit format."""
+
+    id: str
+    repository: str
+    author_email: str
+    author_name: str
+    message: str
+    timestamp: str
+    patch_set: list[Any]
+
+
+class P4DepotPath:
+    """Encapsulates Perforce depot path logic."""
+
+    def __init__(self, path: str):
+        """
+        Initialize depot path.
+
+        Args:
+            path: Depot path (e.g., //depot or //depot/project)
+        """
+        self.path = path
+
+    def depot_name(self) -> str:
+        """
+        Extract depot name from path.
+
+        Returns:
+            Depot name (e.g., "depot" from "//depot/project")
+        """
+        return self.path.strip("/").split("/")[0]
+
 
 class PerforceClient(RepositoryClient, CommitContextClient):
     """
@@ -62,7 +125,7 @@ def __init__(
         self.ssl_fingerprint = metadata.get("ssl_fingerprint")
 
     @contextmanager
-    def _connect(self):
+    def _connect(self) -> Generator[P4]:
         """
         Context manager for P4 connections with automatic cleanup.
 
@@ -241,28 +304,28 @@ def build_depot_path(self, repo: Repository, path: str, stream: str | None = Non
 
         return full_path
 
-    def get_depots(self) -> list[dict[str, Any]]:
+    def get_depots(self) -> Sequence[P4DepotInfo]:
         """
         List all depots accessible to the user.
 
         Uses p4 depots command to display a list of all depots.
         API docs: https://www.perforce.com/manuals/cmdref/Content/CmdRef/p4_depots.html
 
         Returns:
-            List of depot info dictionaries
+            Sequence of depot info dictionaries
         """
         with self._connect() as p4:
             depots = p4.run("depots")
             return [
-                {
-                    "name": depot.get("name"),
-                    "type": depot.get("type"),
-                    "description": depot.get("desc", ""),
-                }
+                P4DepotInfo(
+                    name=str(depot.get("name", "")),
+                    type=str(depot.get("type", "")),
+                    description=str(depot.get("desc", "")),
+                )
                 for depot in depots
             ]
 
-    def get_user(self, username: str) -> dict[str, Any] | None:
+    def get_user(self, username: str) -> P4UserInfo | None:
         """
         Get user information from Perforce.
 
@@ -286,29 +349,87 @@ def get_user(self, username: str) -> dict[str, Any] | None:
                 # Check if user actually exists by verifying Update field is set
                 if not user_info.get("Update"):
                     return None
-                return {
-                    "email": user_info.get("Email", ""),
-                    "full_name": user_info.get("FullName", ""),
-                    "username": user_info.get("User", username),
-                }
+                return P4UserInfo(
+                    email=str(user_info.get("Email", "")),
+                    full_name=str(user_info.get("FullName", "")),
+                    username=str(user_info.get("User", username)),
+                )
             # User not found - return None (not an error condition)
             return None
 
     def get_changes(
-        self, depot_path: str, max_changes: int = 20, start_cl: str | None = None
-    ) -> list[dict[str, Any]]:
+        self,
+        depot_path: str,
+        max_changes: int = 20,
+        start_cl: int | None = None,
+        end_cl: int | None = None,
+    ) -> Sequence[P4ChangeInfo]:
         """
         Get changelists for a depot path.
 
+        Uses p4 changes command to list changelists.
+        API docs: https://www.perforce.com/manuals/cmdref/Content/CmdRef/p4_changes.html
+
         Args:
             depot_path: Depot path (e.g., //depot/main/...)
-            max_changes: Maximum number of changes to return
-            start_cl: Starting changelist number
+            max_changes: Maximum number of changes to return when start_cl/end_cl not specified
+            start_cl: Starting changelist number (exclusive) - returns changes > start_cl. Must be int.
+            end_cl: Ending changelist number (inclusive) - returns changes <= end_cl. Must be int.
 
         Returns:
-            List of changelist dictionaries
+            Sequence of changelist dictionaries in range (start_cl, end_cl]
+
+        Raises:
+            TypeError: If start_cl or end_cl are not integers
         """
-        return []
+        with self._connect() as p4:
+            # Validate types - changelists must be integers
+            if start_cl is not None and not isinstance(start_cl, int):
+                raise TypeError(
+                    f"start_cl must be an integer or None, got {type(start_cl).__name__}"
+                )
+            if end_cl is not None and not isinstance(end_cl, int):
+                raise TypeError(f"end_cl must be an integer or None, got {type(end_cl).__name__}")
+
+            start_cl_num = start_cl
+            end_cl_num = end_cl
+
+            # Calculate how many changes to fetch based on range
+            if start_cl_num is not None and end_cl_num is not None:
+                # Fetch enough to cover the range, adding buffer for safety
+                range_size = abs(end_cl_num - start_cl_num) + DEFAULT_REVISION_RANGE
+                fetch_limit = max(range_size, max_changes)
+            else:
+                fetch_limit = max_changes
+
+            args = ["-m", str(fetch_limit), "-l"]
+
+            # P4 -e flag: return changes at or before specified changelist (upper bound)
+            # Use it for end_cl (inclusive upper bound)
+            if end_cl_num is not None:
+                args.extend(["-e", str(end_cl_num)])
+
+            args.append(depot_path)
+
+            changes = p4.run("changes", *args)
+
+            # Client-side filter for start_cl (exclusive lower bound)
+            # Filter out changes <= start_cl to get changes > start_cl
+            if start_cl_num is not None:
+                changes = [
+                    c for c in changes if c.get("change") and int(c["change"]) > start_cl_num
+                ]
+
+            return [
+                P4ChangeInfo(
+                    change=str(change.get("change", "")),
+                    user=str(change.get("user", "")),
+                    client=str(change.get("client", "")),
+                    time=str(change.get("time", "")),
+                    desc=str(change.get("desc", "")),
+                )
+                for change in changes
+            ]
 
     def get_blame_for_files(
         self, files: Sequence[SourceLineInfo], extra: dict[str, Any]
@@ -331,7 +452,7 @@ def get_file(
     ) -> str:
         """
         Get file contents from Perforce depot.
-        Required by abstract base class but not used (CODEOWNERS feature removed).
+        Required by abstract base class but not used (CODEOWNERS).
         """
         raise NotImplementedError("get_file is not supported for Perforce")
 

src/sentry/integrations/perforce/integration.py

@@ -152,11 +152,11 @@ class PerforceInstallationForm(forms.Form):
         assume_scheme="https",
     )
 
-    def clean_p4port(self):
+    def clean_p4port(self) -> str:
         """Strip off trailing / and whitespace from p4port"""
         return self.cleaned_data["p4port"].strip().rstrip("/")
 
-    def clean_web_url(self):
+    def clean_web_url(self) -> str:
         """Strip off trailing / from web_url"""
         web_url = self.cleaned_data.get("web_url", "")
         if web_url:
@@ -541,7 +541,7 @@ class PerforceIntegrationProvider(IntegrationProvider):
     )
     requires_feature_flag = True
 
-    def get_pipeline_views(self) -> Sequence[PipelineView]:
+    def get_pipeline_views(self) -> Sequence[PipelineView[IntegrationPipeline]]:
         """Get pipeline views for installation flow."""
         return [PerforceInstallationView()]