feat: add branch protection call to client

.github/CODEOWNERS

@@ -7,4 +7,4 @@
 # the repo. Unless a later match takes precedence,
 # @global-owner1 and @global-owner2 will be requested for
 # review when someone opens a pull request.
-*       @jbristow @zprobst @ccloes @angelosantos4 @rreddy2
+*       @jbristow @zprobst @ccloes @angelosantos4 @rreddy2 @AGhafaryy

nodestream_github/client/githubclient.py

@@ -617,3 +617,38 @@ async def fetch_teams_for_repo(self, *, owner_login: str, repo_name: str):
                 yield team
         except httpx.HTTPError as e:
             _fetch_problem(f"teams for repo {owner_login}/{repo_name}", e)
+
+    async def fetch_branch_protection(
+        self,
+        *,
+        owner_login: str,
+        repo_name: str,
+        branch: str,
+    ) -> types.BranchProtection | None:
+        """Fetches the branch protection for a given branch.
+
+        https://docs.github.com/en/[email protected]/rest/branches/branch-protection?apiVersion=2022-11-28#get-branch-protection
+        """
+
+        try:
+            return await self._get_item(
+                f"repos/{owner_login}/{repo_name}/branches/{branch}/protection"
+            )
+        except httpx.HTTPError as e:
+            match e:
+                case httpx.HTTPStatusError(response=response) if (
+                    response.status_code == 404
+                ):
+                    logger.info(
+                        "Branch protection not found for branch %s on repo %s/%s",
+                        branch,
+                        owner_login,
+                        repo_name,
+                    )
+                case _:
+                    _fetch_problem(
+                        f"branch protection for branch {branch} on "
+                        f"repo {owner_login}/{repo_name}",
+                        e,
+                    )
+            return None

nodestream_github/types/__init__.py

@@ -1,4 +1,5 @@
 from .github import (
+    BranchProtection,
     GithubAuditLog,
     GithubOrg,
     GithubOrgSummary,
@@ -19,6 +20,7 @@
 from .httpx import HeaderTypes, PrimitiveData, QueryParamTypes
 
 __all__ = [
+    "BranchProtection",
     "GithubAuditLog",
     "GithubOrg",
     "GithubOrgSummary",

nodestream_github/types/github.py

@@ -22,3 +22,5 @@
 
 SimplifiedRepo: TypeAlias = JSONObject
 SimplifiedUser: TypeAlias = JSONObject
+
+BranchProtection: TypeAlias = JSONObject

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "nodestream-plugin-github"
-version = "0.14.3"
+version = "0.14.4"
 description = ""
 authors = [
     "Jon Bristow <[email protected]>",
@@ -17,17 +17,18 @@ nodestream = "^0.14.14"
 limits = "^5.5.0"
 tenacity = "^9.0.0"
 httpx = ">=0.27,<0.28"
-freezegun = "^1.5.5"
 
 [tool.poetry.group.dev.dependencies]
-ruff = "^0.13.0"
-black = "^25.1.0"
-isort = "^6.0.0"
+black = "^25.9.0"
+freezegun = "^1.5.5"
+isort = "^7.0.0"
 pytest = "^8.4.1"
-pytest-asyncio = "^1.1.0"
-pytest-httpx = "^0.34.0"
+pytest-asyncio = "^1.2.0"
 pytest-cov = "^7.0.0"
 pytest-github-actions-annotate-failures = "^0.3.0"
+pytest-httpx = "^0.34.0"
+pytest-mock = "^3.15.1"
+ruff = "^0.14.2"
 
 [build-system]
 requires = ["poetry-core"]

tests/mocks/githubrest.py

@@ -1,5 +1,5 @@
 # noinspection PyProtectedMember
-from typing import Any, Optional
+from typing import Any
 
 from pytest_httpx import HTTPXMock
 
@@ -192,7 +192,7 @@ def get_repos_for_user(
             **kwargs,
         )
 
-    def get_enterprise_audit_logs(self, *, search_phrase: Optional[str], **kwargs: Any):
+    def get_enterprise_audit_logs(self, *, search_phrase: str | None, **kwargs: Any):
         url = f"{self.base_url}/enterprises/test-enterprise/audit-log"
         url += f"?per_page={self.per_page}"
         if search_phrase:

tests/test_branchprotection.py

@@ -0,0 +1,98 @@
+import httpx
+import pytest
+from pytest_httpx import HTTPXMock
+from pytest_mock import MockerFixture
+
+from nodestream_github.client import GithubRestApiClient
+from nodestream_github.client import githubclient as githubclient
+from tests.mocks.githubrest import DEFAULT_BASE_URL, DEFAULT_HOSTNAME
+
+
+@pytest.mark.asyncio
+async def test_fetch_branch_protection(httpx_mock: HTTPXMock):
+    client = GithubRestApiClient(
+        auth_token="test-auth-token",
+        github_hostname=DEFAULT_HOSTNAME,
+        user_agent="test-user-agent",
+    )
+
+    httpx_mock.add_response(
+        url=f"{DEFAULT_BASE_URL}/repos/octocat/Hello-World/branches/main/protection",
+        json={"enabled": True},
+    )
+
+    result = await client.fetch_branch_protection(
+        owner_login="octocat",
+        repo_name="Hello-World",
+        branch="main",
+    )
+
+    assert result == {"enabled": True}
+
+
+@pytest.mark.asyncio
+async def test_fetch_branch_protection_404(
+    httpx_mock: HTTPXMock, mocker: MockerFixture
+):
+    client = GithubRestApiClient(
+        auth_token="test-auth-token",
+        github_hostname=DEFAULT_HOSTNAME,
+        user_agent="test-user-agent",
+    )
+
+    httpx_mock.add_response(
+        url=f"{DEFAULT_BASE_URL}/repos/octocat/Hello-World/branches/main/protection",
+        status_code=httpx.codes.NOT_FOUND,
+        json={
+            "documentation_url": "https://docs.github.com/[email protected]/rest",
+            "message": "Not Found",
+        },
+    )
+    log_info = mocker.spy(githubclient.logger, "info")
+
+    result = await client.fetch_branch_protection(
+        owner_login="octocat",
+        repo_name="Hello-World",
+        branch="main",
+    )
+
+    assert result is None
+    log_info.assert_called_once_with(
+        "Branch protection not found for branch %s on repo %s/%s",
+        "main",
+        "octocat",
+        "Hello-World",
+    )
+
+
+@pytest.mark.asyncio
+async def test_fetch_branch_protection_503(
+    httpx_mock: HTTPXMock, mocker: MockerFixture
+):
+    client = GithubRestApiClient(
+        auth_token="test-auth-token",
+        github_hostname=DEFAULT_HOSTNAME,
+        user_agent="test-user-agent",
+    )
+
+    httpx_mock.add_response(
+        url=f"{DEFAULT_BASE_URL}/repos/octocat/Hello-World/branches/main/protection",
+        status_code=httpx.codes.SERVICE_UNAVAILABLE,
+    )
+    log_warning = mocker.spy(githubclient.logger, "warning")
+
+    result = await client.fetch_branch_protection(
+        owner_login="octocat",
+        repo_name="Hello-World",
+        branch="main",
+    )
+
+    assert result is None
+    log_warning.assert_called_once_with(
+        "%s %s - %s%s",
+        503,
+        "Service Unavailable",
+        "/api/v3/repos/octocat/Hello-World/branches/main/protection",
+        "",
+        stacklevel=2,
+    )