From 5033fa26b3a1908e69d2d027c853d90ad08bc452 Mon Sep 17 00:00:00 2001 From: mementorc Date: Fri, 19 Jun 2026 10:22:28 -0500 Subject: [PATCH 1/3] =?UTF-8?q?feat(github):=20add=20GHAS=20forensics=20to?= =?UTF-8?q?ols=20=E2=80=94=20rulesets,=20code=20&=20secret=20scanning=20(#?= =?UTF-8?q?186)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds 7 read-only tools so 'mergeable: true, merge_state: blocked' PRs whose blocker lives in a repository ruleset or a code-scanning requirement can be root-caused from MCP git alone (classic branch protection doesn't expose these). Rulesets: github_list_rulesets, github_get_ruleset, github_get_branch_rules. Code scanning: github_list_code_scanning_alerts, github_list_code_scanning_analyses, github_get_code_scanning_default_setup. Secret scanning: github_list_secret_scanning_alerts. Closes #186 Co-Authored-By: Claude Opus 4.8 --- README.md | 4 +- src/mcp_server_git/github/api.py | 1 + src/mcp_server_git/github/models.py | 71 +++ src/mcp_server_git/github/scanning.py | 361 ++++++++++++++ src/mcp_server_git/lean/registry_github.py | 66 +++ .../unit/github/test_github_ghas_forensics.py | 458 ++++++++++++++++++ 6 files changed, 959 insertions(+), 2 deletions(-) create mode 100644 src/mcp_server_git/github/scanning.py create mode 100644 tests/unit/github/test_github_ghas_forensics.py diff --git a/README.md b/README.md index a9186ddc..3fe7da6d 100644 --- a/README.md +++ b/README.md @@ -137,7 +137,7 @@ The server provides Azure DevOps integration for monitoring and analyzing Azure ### Lean MCP Interface (Context-Optimized) -The server provides an alternative **lean interface** that reduces context consumption by ~97% (from ~30k tokens to ~900 tokens). Instead of exposing all 56 tools upfront, it uses a 3-meta-tool pattern: +The server provides an alternative **lean interface** that reduces context consumption by ~97% (from ~30k tokens to ~900 tokens). Instead of exposing all 63 tools upfront, it uses a 3-meta-tool pattern: | Meta-Tool | Purpose | |-----------|---------| @@ -145,7 +145,7 @@ The server provides an alternative **lean interface** that reduces context consu | `get_tool_spec(tool_name)` | Get full schema for a specific tool on-demand | | `execute_tool(tool_name, params)` | Execute any tool dynamically | -**Tool Coverage**: All 55 tools remain accessible (27 git, 25 GitHub, 3 Azure DevOps). +**Tool Coverage**: All 62 tools remain accessible (27 git, 32 GitHub, 3 Azure DevOps). **Usage Example**: ```python diff --git a/src/mcp_server_git/github/api.py b/src/mcp_server_git/github/api.py index e8bd22ca..cc602f6a 100644 --- a/src/mcp_server_git/github/api.py +++ b/src/mcp_server_git/github/api.py @@ -13,6 +13,7 @@ from .release_assets import * # noqa: F401,F403 from .releases import * # noqa: F401,F403 from .repos import * # noqa: F401,F403 +from .scanning import * # noqa: F401,F403 from .security import * # noqa: F401,F403 from .workflows import * # noqa: F401,F403 diff --git a/src/mcp_server_git/github/models.py b/src/mcp_server_git/github/models.py index 41da750c..b5df829e 100644 --- a/src/mcp_server_git/github/models.py +++ b/src/mcp_server_git/github/models.py @@ -980,3 +980,74 @@ def validate_org_name(cls, v: str | None) -> str | None: "can only contain alphanumeric characters and hyphens" ) return v + + +# ============================================================================ +# GHAS Forensics Models (Issue #186) +# ============================================================================ + + +class GitHubListRulesets(BaseModel): + """Model for listing rulesets defined on a repository.""" + + repo_owner: str + repo_name: str + + +class GitHubGetRuleset(BaseModel): + """Model for fetching a specific ruleset by ID. + + Returns full ruleset config including required_status_checks, + code_scanning rules, and bypass actors. + """ + + repo_owner: str + repo_name: str + ruleset_id: int # Numeric ruleset ID (from github_list_rulesets) + + +class GitHubGetBranchRules(BaseModel): + """Model for listing all rules that apply to a branch. + + Returns both classic branch-protection rules and ruleset-based rules + active for the given branch ref. + """ + + repo_owner: str + repo_name: str + branch: str # Branch name (e.g. 'main') or pattern + + +class GitHubListCodeScanningAlerts(BaseModel): + """Model for listing code scanning alerts with optional filters.""" + + repo_owner: str + repo_name: str + state: str | None = None # 'open', 'closed', 'dismissed', or 'fixed' + severity: str | None = None # 'critical', 'high', 'medium', 'low', 'warning', 'note', 'error' + tool_name: str | None = None # Code-scanning tool (e.g. 'CodeQL') + ref: str | None = None # Branch name or refs/pull/N/head + + +class GitHubListCodeScanningAnalyses(BaseModel): + """Model for listing code scanning analyses for a repository.""" + + repo_owner: str + repo_name: str + ref: str | None = None # Branch name or refs/pull/N/head + tool_name: str | None = None # Code-scanning tool (e.g. 'CodeQL') + + +class GitHubGetCodeScanningDefaultSetup(BaseModel): + """Model for fetching the default-setup configuration for code scanning.""" + + repo_owner: str + repo_name: str + + +class GitHubListSecretScanningAlerts(BaseModel): + """Model for listing secret scanning alerts for a repository.""" + + repo_owner: str + repo_name: str + state: str | None = None # 'open' or 'resolved' diff --git a/src/mcp_server_git/github/scanning.py b/src/mcp_server_git/github/scanning.py new file mode 100644 index 00000000..06c78194 --- /dev/null +++ b/src/mcp_server_git/github/scanning.py @@ -0,0 +1,361 @@ +"""GitHub GHAS forensics — rulesets, code scanning, and secret scanning.""" + +from __future__ import annotations + +import logging +from typing import Any + +from mcp_server_git.github.client import github_client_context + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _build_params(**kwargs: Any) -> dict[str, str]: + """Build query-param dict omitting None values.""" + return {k: str(v) for k, v in kwargs.items() if v is not None} + + +# --------------------------------------------------------------------------- +# Rulesets +# --------------------------------------------------------------------------- + + +async def github_list_rulesets( + repo_owner: str, + repo_name: str, +) -> list[dict[str, Any]] | str: + """List rulesets defined on a repository.""" + logger.debug("Getting rulesets for %s/%s", repo_owner, repo_name) + + try: + async with github_client_context() as client: + response = await client.get( + f"/repos/{repo_owner}/{repo_name}/rulesets" + ) + + if response.status == 200: + return await response.json() + elif response.status == 404: + return f"❌ Repository not found: {repo_owner}/{repo_name}" + else: + error_text = await response.text() + return f"❌ Failed to list rulesets: {response.status} - {error_text}" + + except ValueError as auth_error: + logger.error("Authentication error listing rulesets: %s", auth_error) + return f"❌ {auth_error}" + except ConnectionError as conn_error: + logger.error("Connection error listing rulesets: %s", conn_error) + return f"❌ Network connection failed: {conn_error}" + except Exception as exc: + logger.error("Unexpected error listing rulesets: %s", exc, exc_info=True) + return f"❌ Error listing rulesets: {exc}" + + +async def github_get_ruleset( + repo_owner: str, + repo_name: str, + ruleset_id: int, +) -> dict[str, Any] | str: + """Get a specific ruleset by ID (includes required_status_checks, code_scanning, bypass actors).""" + logger.debug( + "Getting ruleset %d for %s/%s", ruleset_id, repo_owner, repo_name + ) + + try: + async with github_client_context() as client: + response = await client.get( + f"/repos/{repo_owner}/{repo_name}/rulesets/{ruleset_id}" + ) + + if response.status == 200: + return await response.json() + elif response.status == 404: + return ( + f"❌ Ruleset {ruleset_id} not found for {repo_owner}/{repo_name}" + ) + else: + error_text = await response.text() + return f"❌ Failed to get ruleset: {response.status} - {error_text}" + + except ValueError as auth_error: + logger.error("Authentication error getting ruleset: %s", auth_error) + return f"❌ {auth_error}" + except ConnectionError as conn_error: + logger.error("Connection error getting ruleset: %s", conn_error) + return f"❌ Network connection failed: {conn_error}" + except Exception as exc: + logger.error("Unexpected error getting ruleset: %s", exc, exc_info=True) + return f"❌ Error getting ruleset: {exc}" + + +async def github_get_branch_rules( + repo_owner: str, + repo_name: str, + branch: str, +) -> list[dict[str, Any]] | str: + """Get all rules (classic + ruleset-based) that apply to a branch.""" + logger.debug( + "Getting branch rules for %s/%s#%s", repo_owner, repo_name, branch + ) + + try: + async with github_client_context() as client: + response = await client.get( + f"/repos/{repo_owner}/{repo_name}/rules/branches/{branch}" + ) + + if response.status == 200: + return await response.json() + elif response.status == 404: + return ( + f"❌ Branch or repository not found: {repo_owner}/{repo_name}#{branch}" + ) + else: + error_text = await response.text() + return f"❌ Failed to get branch rules: {response.status} - {error_text}" + + except ValueError as auth_error: + logger.error("Authentication error getting branch rules: %s", auth_error) + return f"❌ {auth_error}" + except ConnectionError as conn_error: + logger.error("Connection error getting branch rules: %s", conn_error) + return f"❌ Network connection failed: {conn_error}" + except Exception as exc: + logger.error("Unexpected error getting branch rules: %s", exc, exc_info=True) + return f"❌ Error getting branch rules: {exc}" + + +# --------------------------------------------------------------------------- +# Code Scanning +# --------------------------------------------------------------------------- + + +async def github_list_code_scanning_alerts( + repo_owner: str, + repo_name: str, + state: str | None = None, + severity: str | None = None, + tool_name: str | None = None, + ref: str | None = None, +) -> list[dict[str, Any]] | str: + """List code scanning alerts with optional filters. + + Args: + state: Filter by alert state ('open', 'closed', 'dismissed', 'fixed'). + severity: Filter by severity ('critical', 'high', 'medium', 'low', 'warning', 'note', 'error'). + tool_name: Filter by code-scanning tool name (e.g. 'CodeQL'). + ref: Filter by Git ref (branch name or refs/pull/N/head). + """ + logger.debug( + "Listing code scanning alerts for %s/%s", repo_owner, repo_name + ) + + params = _build_params( + state=state, severity=severity, tool_name=tool_name, ref=ref + ) + + try: + async with github_client_context() as client: + response = await client.get( + f"/repos/{repo_owner}/{repo_name}/code-scanning/alerts", + params=params if params else None, + ) + + if response.status == 200: + return await response.json() + elif response.status == 404: + return ( + f"❌ Repository not found or code scanning not enabled: " + f"{repo_owner}/{repo_name}" + ) + else: + error_text = await response.text() + return ( + f"❌ Failed to list code scanning alerts: " + f"{response.status} - {error_text}" + ) + + except ValueError as auth_error: + logger.error( + "Authentication error listing code scanning alerts: %s", auth_error + ) + return f"❌ {auth_error}" + except ConnectionError as conn_error: + logger.error( + "Connection error listing code scanning alerts: %s", conn_error + ) + return f"❌ Network connection failed: {conn_error}" + except Exception as exc: + logger.error( + "Unexpected error listing code scanning alerts: %s", exc, exc_info=True + ) + return f"❌ Error listing code scanning alerts: {exc}" + + +async def github_list_code_scanning_analyses( + repo_owner: str, + repo_name: str, + ref: str | None = None, + tool_name: str | None = None, +) -> list[dict[str, Any]] | str: + """List code scanning analyses for a repository. + + Args: + ref: Git ref to filter by (branch name or refs/pull/N/head). + tool_name: Code-scanning tool name to filter by (e.g. 'CodeQL'). + """ + logger.debug( + "Listing code scanning analyses for %s/%s", repo_owner, repo_name + ) + + params = _build_params(ref=ref, tool_name=tool_name) + + try: + async with github_client_context() as client: + response = await client.get( + f"/repos/{repo_owner}/{repo_name}/code-scanning/analyses", + params=params if params else None, + ) + + if response.status == 200: + return await response.json() + elif response.status == 404: + return ( + f"❌ Repository not found or code scanning not enabled: " + f"{repo_owner}/{repo_name}" + ) + else: + error_text = await response.text() + return ( + f"❌ Failed to list code scanning analyses: " + f"{response.status} - {error_text}" + ) + + except ValueError as auth_error: + logger.error( + "Authentication error listing code scanning analyses: %s", auth_error + ) + return f"❌ {auth_error}" + except ConnectionError as conn_error: + logger.error( + "Connection error listing code scanning analyses: %s", conn_error + ) + return f"❌ Network connection failed: {conn_error}" + except Exception as exc: + logger.error( + "Unexpected error listing code scanning analyses: %s", exc, exc_info=True + ) + return f"❌ Error listing code scanning analyses: {exc}" + + +async def github_get_code_scanning_default_setup( + repo_owner: str, + repo_name: str, +) -> dict[str, Any] | str: + """Get the default-setup configuration for code scanning.""" + logger.debug( + "Getting code scanning default setup for %s/%s", repo_owner, repo_name + ) + + try: + async with github_client_context() as client: + response = await client.get( + f"/repos/{repo_owner}/{repo_name}/code-scanning/default-setup" + ) + + if response.status == 200: + return await response.json() + elif response.status == 404: + return ( + f"❌ Repository not found or code scanning not available: " + f"{repo_owner}/{repo_name}" + ) + else: + error_text = await response.text() + return ( + f"❌ Failed to get code scanning default setup: " + f"{response.status} - {error_text}" + ) + + except ValueError as auth_error: + logger.error( + "Authentication error getting code scanning default setup: %s", auth_error + ) + return f"❌ {auth_error}" + except ConnectionError as conn_error: + logger.error( + "Connection error getting code scanning default setup: %s", conn_error + ) + return f"❌ Network connection failed: {conn_error}" + except Exception as exc: + logger.error( + "Unexpected error getting code scanning default setup: %s", + exc, + exc_info=True, + ) + return f"❌ Error getting code scanning default setup: {exc}" + + +# --------------------------------------------------------------------------- +# Secret Scanning +# --------------------------------------------------------------------------- + + +async def github_list_secret_scanning_alerts( + repo_owner: str, + repo_name: str, + state: str | None = None, +) -> list[dict[str, Any]] | str: + """List secret scanning alerts for a repository. + + Args: + state: Filter by alert state ('open' or 'resolved'). + """ + logger.debug( + "Listing secret scanning alerts for %s/%s", repo_owner, repo_name + ) + + params = _build_params(state=state) + + try: + async with github_client_context() as client: + response = await client.get( + f"/repos/{repo_owner}/{repo_name}/secret-scanning/alerts", + params=params if params else None, + ) + + if response.status == 200: + return await response.json() + elif response.status == 404: + return ( + f"❌ Repository not found or secret scanning not enabled: " + f"{repo_owner}/{repo_name}" + ) + else: + error_text = await response.text() + return ( + f"❌ Failed to list secret scanning alerts: " + f"{response.status} - {error_text}" + ) + + except ValueError as auth_error: + logger.error( + "Authentication error listing secret scanning alerts: %s", auth_error + ) + return f"❌ {auth_error}" + except ConnectionError as conn_error: + logger.error( + "Connection error listing secret scanning alerts: %s", conn_error + ) + return f"❌ Network connection failed: {conn_error}" + except Exception as exc: + logger.error( + "Unexpected error listing secret scanning alerts: %s", exc, exc_info=True + ) + return f"❌ Error listing secret scanning alerts: {exc}" diff --git a/src/mcp_server_git/lean/registry_github.py b/src/mcp_server_git/lean/registry_github.py index 8f34957b..5e64a270 100644 --- a/src/mcp_server_git/lean/registry_github.py +++ b/src/mcp_server_git/lean/registry_github.py @@ -23,6 +23,8 @@ GitHubGetActionsPermissions, GitHubGetAutomatedSecurityFixes, GitHubGetBranchProtection, + GitHubGetBranchRules, + GitHubGetCodeScanningDefaultSetup, GitHubGetFailingJobs, GitHubGetIssue, GitHubGetJobLogs, @@ -33,14 +35,19 @@ GitHubGetRelease, GitHubGetRequiredSignatures, GitHubGetRepoSettings, + GitHubGetRuleset, GitHubGetSecurityAnalysis, GitHubGetVulnerabilityAlerts, GitHubGetWorkflowPermissions, GitHubGetWorkflowRun, + GitHubListCodeScanningAlerts, + GitHubListCodeScanningAnalyses, GitHubListIssues, GitHubListPullRequests, GitHubListReleaseAssets, GitHubListReleases, + GitHubListRulesets, + GitHubListSecretScanningAlerts, GitHubListWorkflowRuns, GitHubSearchIssues, GitHubUpdateActionsPermissions, @@ -509,6 +516,65 @@ def _register_github_tools(interface: Any, github_service: Any): domain="github", complexity="comprehensive", ), + # GHAS Forensics — Rulesets (Issue #186) + ToolDefinition( + name="github_list_rulesets", + implementation=github_ops.github_list_rulesets, + description="List rulesets defined on a repository (complements classic branch protection)", + schema=GitHubListRulesets.model_json_schema(), + domain="github", + complexity="focused", + ), + ToolDefinition( + name="github_get_ruleset", + implementation=github_ops.github_get_ruleset, + description="Get a specific ruleset by ID including required_status_checks, code_scanning rules, and bypass actors", + schema=GitHubGetRuleset.model_json_schema(), + domain="github", + complexity="focused", + ), + ToolDefinition( + name="github_get_branch_rules", + implementation=github_ops.github_get_branch_rules, + description="Get all rules (classic + ruleset-based) that apply to a branch", + schema=GitHubGetBranchRules.model_json_schema(), + domain="github", + complexity="focused", + ), + # GHAS Forensics — Code Scanning (Issue #186) + ToolDefinition( + name="github_list_code_scanning_alerts", + implementation=github_ops.github_list_code_scanning_alerts, + description="List code scanning alerts with optional state/severity/tool/ref filters", + schema=GitHubListCodeScanningAlerts.model_json_schema(), + domain="github", + complexity="focused", + ), + ToolDefinition( + name="github_list_code_scanning_analyses", + implementation=github_ops.github_list_code_scanning_analyses, + description="List code scanning analyses for a repository, filterable by ref and tool", + schema=GitHubListCodeScanningAnalyses.model_json_schema(), + domain="github", + complexity="focused", + ), + ToolDefinition( + name="github_get_code_scanning_default_setup", + implementation=github_ops.github_get_code_scanning_default_setup, + description="Get the default-setup configuration for code scanning (languages, query suite, schedule)", + schema=GitHubGetCodeScanningDefaultSetup.model_json_schema(), + domain="github", + complexity="focused", + ), + # GHAS Forensics — Secret Scanning (Issue #186) + ToolDefinition( + name="github_list_secret_scanning_alerts", + implementation=github_ops.github_list_secret_scanning_alerts, + description="List secret scanning alerts for a repository with optional state filter", + schema=GitHubListSecretScanningAlerts.model_json_schema(), + domain="github", + complexity="focused", + ), # Release Management Tools ToolDefinition( name="github_create_release", diff --git a/tests/unit/github/test_github_ghas_forensics.py b/tests/unit/github/test_github_ghas_forensics.py new file mode 100644 index 00000000..054b2878 --- /dev/null +++ b/tests/unit/github/test_github_ghas_forensics.py @@ -0,0 +1,458 @@ +""" +Unit tests for GitHub GHAS forensics functions (Issue #186). + +Covers 7 read-only tools: + Rulesets: github_list_rulesets, github_get_ruleset, github_get_branch_rules + Code scanning: github_list_code_scanning_alerts, github_list_code_scanning_analyses, + github_get_code_scanning_default_setup + Secret scanning: github_list_secret_scanning_alerts +""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from src.mcp_server_git.github.api import ( + github_get_branch_rules, + github_get_code_scanning_default_setup, + github_get_ruleset, + github_list_code_scanning_alerts, + github_list_code_scanning_analyses, + github_list_rulesets, + github_list_secret_scanning_alerts, +) + +_OWNER = "myorg" +_REPO = "myrepo" + +_PATCH_CTX = "src.mcp_server_git.github.scanning.github_client_context" + + +def _mock_client(method: str, status: int, json_body=None, text_body: str = ""): + """Create a mock HTTP client returning the given status and body.""" + mock_response = AsyncMock() + mock_response.status = status + if json_body is not None: + mock_response.json = AsyncMock(return_value=json_body) + mock_response.text = AsyncMock(return_value=text_body) + + mock_client = MagicMock() + setattr(mock_client, method, AsyncMock(return_value=mock_response)) + return mock_client, mock_response + + +# =========================================================================== +# github_list_rulesets +# =========================================================================== + + +class TestGithubListRulesets: + """Tests for github_list_rulesets.""" + + @pytest.mark.asyncio + async def test_list_rulesets_returns_list_when_status_200(self): + """GET 200 returns the JSON list of rulesets.""" + rulesets = [{"id": 1, "name": "default"}] + mock_client, _ = _mock_client("get", 200, json_body=rulesets) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_list_rulesets( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert result == rulesets + mock_client.get.assert_called_once_with( + f"/repos/{_OWNER}/{_REPO}/rulesets" + ) + + @pytest.mark.asyncio + async def test_list_rulesets_returns_not_found_when_status_404(self): + """GET 404 returns a structured not-found string.""" + mock_client, _ = _mock_client("get", 404, text_body="Not Found") + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_list_rulesets( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert "❌" in result + assert "not found" in result.lower() + assert f"{_OWNER}/{_REPO}" in result + + +# =========================================================================== +# github_get_ruleset +# =========================================================================== + + +class TestGithubGetRuleset: + """Tests for github_get_ruleset.""" + + @pytest.mark.asyncio + async def test_get_ruleset_returns_dict_when_status_200(self): + """GET 200 returns the full ruleset dict.""" + ruleset = {"id": 42, "name": "ci-required", "rules": []} + mock_client, _ = _mock_client("get", 200, json_body=ruleset) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_get_ruleset( + repo_owner=_OWNER, repo_name=_REPO, ruleset_id=42 + ) + + assert result == ruleset + mock_client.get.assert_called_once_with( + f"/repos/{_OWNER}/{_REPO}/rulesets/42" + ) + + @pytest.mark.asyncio + async def test_get_ruleset_returns_not_found_when_status_404(self): + """GET 404 returns a structured not-found string containing the ruleset ID.""" + mock_client, _ = _mock_client("get", 404, text_body="Not Found") + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_get_ruleset( + repo_owner=_OWNER, repo_name=_REPO, ruleset_id=42 + ) + + assert "❌" in result + assert "42" in result + assert f"{_OWNER}/{_REPO}" in result + + +# =========================================================================== +# github_get_branch_rules +# =========================================================================== + + +class TestGithubGetBranchRules: + """Tests for github_get_branch_rules.""" + + @pytest.mark.asyncio + async def test_get_branch_rules_returns_list_when_status_200(self): + """GET 200 returns the list of rules for the branch.""" + rules = [{"type": "required_signatures"}, {"type": "code_scanning"}] + mock_client, _ = _mock_client("get", 200, json_body=rules) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_get_branch_rules( + repo_owner=_OWNER, repo_name=_REPO, branch="main" + ) + + assert result == rules + mock_client.get.assert_called_once_with( + f"/repos/{_OWNER}/{_REPO}/rules/branches/main" + ) + + @pytest.mark.asyncio + async def test_get_branch_rules_returns_not_found_when_status_404(self): + """GET 404 returns a not-found string mentioning branch and repo.""" + mock_client, _ = _mock_client("get", 404, text_body="Not Found") + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_get_branch_rules( + repo_owner=_OWNER, repo_name=_REPO, branch="main" + ) + + assert "❌" in result + assert f"{_OWNER}/{_REPO}#main" in result + + +# =========================================================================== +# github_list_code_scanning_alerts +# =========================================================================== + + +class TestGithubListCodeScanningAlerts: + """Tests for github_list_code_scanning_alerts.""" + + @pytest.mark.asyncio + async def test_list_alerts_returns_list_when_status_200(self): + """GET 200 returns the alert list.""" + alerts = [{"number": 1, "state": "open", "rule": {"severity": "critical"}}] + mock_client, _ = _mock_client("get", 200, json_body=alerts) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_list_code_scanning_alerts( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert result == alerts + mock_client.get.assert_called_once_with( + f"/repos/{_OWNER}/{_REPO}/code-scanning/alerts", + params=None, + ) + + @pytest.mark.asyncio + async def test_list_alerts_passes_filters_as_query_params(self): + """Provided filter args are forwarded as query params; None args are omitted.""" + mock_client, _ = _mock_client("get", 200, json_body=[]) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + await github_list_code_scanning_alerts( + repo_owner=_OWNER, + repo_name=_REPO, + state="open", + severity="critical", + tool_name="CodeQL", + ref="refs/pull/207/head", + ) + + call_kwargs = mock_client.get.call_args + params = call_kwargs[1]["params"] + assert params["state"] == "open" + assert params["severity"] == "critical" + assert params["tool_name"] == "CodeQL" + assert params["ref"] == "refs/pull/207/head" + + @pytest.mark.asyncio + async def test_list_alerts_omits_none_filters(self): + """None filter args must NOT appear in query params.""" + mock_client, _ = _mock_client("get", 200, json_body=[]) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + await github_list_code_scanning_alerts( + repo_owner=_OWNER, + repo_name=_REPO, + state="open", + severity=None, + tool_name=None, + ref=None, + ) + + call_kwargs = mock_client.get.call_args + params = call_kwargs[1]["params"] + assert "severity" not in params + assert "tool_name" not in params + assert "ref" not in params + + @pytest.mark.asyncio + async def test_list_alerts_returns_not_found_when_status_404(self): + """GET 404 returns a not-found/not-enabled string.""" + mock_client, _ = _mock_client("get", 404, text_body="Not Found") + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_list_code_scanning_alerts( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert "❌" in result + assert f"{_OWNER}/{_REPO}" in result + + +# =========================================================================== +# github_list_code_scanning_analyses +# =========================================================================== + + +class TestGithubListCodeScanningAnalyses: + """Tests for github_list_code_scanning_analyses.""" + + @pytest.mark.asyncio + async def test_list_analyses_returns_list_when_status_200(self): + """GET 200 returns the analysis list.""" + analyses = [{"id": 1, "ref": "refs/heads/main", "tool": {"name": "CodeQL"}}] + mock_client, _ = _mock_client("get", 200, json_body=analyses) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_list_code_scanning_analyses( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert result == analyses + mock_client.get.assert_called_once_with( + f"/repos/{_OWNER}/{_REPO}/code-scanning/analyses", + params=None, + ) + + @pytest.mark.asyncio + async def test_list_analyses_passes_filters_as_query_params(self): + """ref and tool_name are forwarded; None values omitted.""" + mock_client, _ = _mock_client("get", 200, json_body=[]) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + await github_list_code_scanning_analyses( + repo_owner=_OWNER, + repo_name=_REPO, + ref="refs/pull/207/head", + tool_name="CodeQL", + ) + + call_kwargs = mock_client.get.call_args + params = call_kwargs[1]["params"] + assert params["ref"] == "refs/pull/207/head" + assert params["tool_name"] == "CodeQL" + + @pytest.mark.asyncio + async def test_list_analyses_omits_none_filters(self): + """None ref and tool_name must not appear in params.""" + mock_client, _ = _mock_client("get", 200, json_body=[]) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + await github_list_code_scanning_analyses( + repo_owner=_OWNER, + repo_name=_REPO, + ref=None, + tool_name=None, + ) + + call_kwargs = mock_client.get.call_args + params = call_kwargs[1]["params"] + assert params is None + + @pytest.mark.asyncio + async def test_list_analyses_returns_not_found_when_status_404(self): + """GET 404 returns a not-found/not-enabled string.""" + mock_client, _ = _mock_client("get", 404, text_body="Not Found") + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_list_code_scanning_analyses( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert "❌" in result + assert f"{_OWNER}/{_REPO}" in result + + +# =========================================================================== +# github_get_code_scanning_default_setup +# =========================================================================== + + +class TestGithubGetCodeScanningDefaultSetup: + """Tests for github_get_code_scanning_default_setup.""" + + @pytest.mark.asyncio + async def test_get_default_setup_returns_dict_when_status_200(self): + """GET 200 returns the default-setup configuration.""" + config = {"state": "configured", "languages": ["python"], "query_suite": "default"} + mock_client, _ = _mock_client("get", 200, json_body=config) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_get_code_scanning_default_setup( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert result == config + mock_client.get.assert_called_once_with( + f"/repos/{_OWNER}/{_REPO}/code-scanning/default-setup" + ) + + @pytest.mark.asyncio + async def test_get_default_setup_returns_not_found_when_status_404(self): + """GET 404 returns a not-found/not-available string.""" + mock_client, _ = _mock_client("get", 404, text_body="Not Found") + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_get_code_scanning_default_setup( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert "❌" in result + assert f"{_OWNER}/{_REPO}" in result + + +# =========================================================================== +# github_list_secret_scanning_alerts +# =========================================================================== + + +class TestGithubListSecretScanningAlerts: + """Tests for github_list_secret_scanning_alerts.""" + + @pytest.mark.asyncio + async def test_list_secret_alerts_returns_list_when_status_200(self): + """GET 200 returns the alert list.""" + alerts = [{"number": 1, "state": "open", "secret_type": "github_personal_access_token"}] + mock_client, _ = _mock_client("get", 200, json_body=alerts) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_list_secret_scanning_alerts( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert result == alerts + mock_client.get.assert_called_once_with( + f"/repos/{_OWNER}/{_REPO}/secret-scanning/alerts", + params=None, + ) + + @pytest.mark.asyncio + async def test_list_secret_alerts_passes_state_filter(self): + """Provided state is forwarded as a query param.""" + mock_client, _ = _mock_client("get", 200, json_body=[]) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + await github_list_secret_scanning_alerts( + repo_owner=_OWNER, repo_name=_REPO, state="open" + ) + + call_kwargs = mock_client.get.call_args + params = call_kwargs[1]["params"] + assert params["state"] == "open" + + @pytest.mark.asyncio + async def test_list_secret_alerts_omits_none_state(self): + """None state must not appear in params.""" + mock_client, _ = _mock_client("get", 200, json_body=[]) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + await github_list_secret_scanning_alerts( + repo_owner=_OWNER, repo_name=_REPO, state=None + ) + + call_kwargs = mock_client.get.call_args + params = call_kwargs[1]["params"] + assert params is None + + @pytest.mark.asyncio + async def test_list_secret_alerts_returns_not_found_when_status_404(self): + """GET 404 returns a not-found/not-enabled string.""" + mock_client, _ = _mock_client("get", 404, text_body="Not Found") + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_list_secret_scanning_alerts( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert "❌" in result + assert f"{_OWNER}/{_REPO}" in result From c03c3694b6d2c639bbb494cddc248c09c5b453db Mon Sep 17 00:00:00 2001 From: mementorc Date: Fri, 19 Jun 2026 11:11:41 -0500 Subject: [PATCH 2/3] =?UTF-8?q?fix(github):=20address=20PR=20#188=20review?= =?UTF-8?q?=20=E2=80=94=20pagination,=20=5F=5Fall=5F=5F,=20enum=20validati?= =?UTF-8?q?on?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add per_page/page to the four GHAS list tools (rulesets, code-scanning alerts/analyses, secret-scanning alerts) — endpoints paginate (30 default, 100 max); previously only the first page was returned. - Add __all__ to scanning.py so wildcard re-export hides logger/_build_params. - Drop empty-string query params in _build_params (keep False/0). - Literal-constrain code-scanning state/severity and secret-scanning state for schema-documented validation. - Add 403 / exception error-path tests; use call_args.kwargs in assertions. Co-Authored-By: Claude Opus 4.8 --- src/mcp_server_git/github/models.py | 20 +- src/mcp_server_git/github/scanning.py | 40 +++- .../unit/github/test_github_ghas_forensics.py | 173 ++++++++++++++++-- 3 files changed, 209 insertions(+), 24 deletions(-) diff --git a/src/mcp_server_git/github/models.py b/src/mcp_server_git/github/models.py index b5df829e..0cd95b14 100644 --- a/src/mcp_server_git/github/models.py +++ b/src/mcp_server_git/github/models.py @@ -2,7 +2,9 @@ import re -from pydantic import BaseModel, field_validator, model_validator +from typing import Literal + +from pydantic import BaseModel, Field, field_validator, model_validator # ============================================================================ # Validation Constants @@ -992,6 +994,8 @@ class GitHubListRulesets(BaseModel): repo_owner: str repo_name: str + per_page: int | None = Field(default=None, ge=1, le=100, description="Results per page (max 100)") + page: int | None = Field(default=None, ge=1, description="Page number") class GitHubGetRuleset(BaseModel): @@ -1023,10 +1027,12 @@ class GitHubListCodeScanningAlerts(BaseModel): repo_owner: str repo_name: str - state: str | None = None # 'open', 'closed', 'dismissed', or 'fixed' - severity: str | None = None # 'critical', 'high', 'medium', 'low', 'warning', 'note', 'error' - tool_name: str | None = None # Code-scanning tool (e.g. 'CodeQL') + state: Literal["open", "closed", "dismissed", "fixed"] | None = None + severity: Literal["critical", "high", "medium", "low", "warning", "note", "error"] | None = None + tool_name: str | None = None # Code-scanning tool (e.g. 'CodeQL') — free-form ref: str | None = None # Branch name or refs/pull/N/head + per_page: int | None = Field(default=None, ge=1, le=100, description="Results per page (max 100)") + page: int | None = Field(default=None, ge=1, description="Page number") class GitHubListCodeScanningAnalyses(BaseModel): @@ -1036,6 +1042,8 @@ class GitHubListCodeScanningAnalyses(BaseModel): repo_name: str ref: str | None = None # Branch name or refs/pull/N/head tool_name: str | None = None # Code-scanning tool (e.g. 'CodeQL') + per_page: int | None = Field(default=None, ge=1, le=100, description="Results per page (max 100)") + page: int | None = Field(default=None, ge=1, description="Page number") class GitHubGetCodeScanningDefaultSetup(BaseModel): @@ -1050,4 +1058,6 @@ class GitHubListSecretScanningAlerts(BaseModel): repo_owner: str repo_name: str - state: str | None = None # 'open' or 'resolved' + state: Literal["open", "resolved"] | None = None + per_page: int | None = Field(default=None, ge=1, le=100, description="Results per page (max 100)") + page: int | None = Field(default=None, ge=1, description="Page number") diff --git a/src/mcp_server_git/github/scanning.py b/src/mcp_server_git/github/scanning.py index 06c78194..bcf6d042 100644 --- a/src/mcp_server_git/github/scanning.py +++ b/src/mcp_server_git/github/scanning.py @@ -9,6 +9,16 @@ logger = logging.getLogger(__name__) +__all__ = [ + "github_list_rulesets", + "github_get_ruleset", + "github_get_branch_rules", + "github_list_code_scanning_alerts", + "github_list_code_scanning_analyses", + "github_get_code_scanning_default_setup", + "github_list_secret_scanning_alerts", +] + # --------------------------------------------------------------------------- # Helpers @@ -16,8 +26,8 @@ def _build_params(**kwargs: Any) -> dict[str, str]: - """Build query-param dict omitting None values.""" - return {k: str(v) for k, v in kwargs.items() if v is not None} + """Build query-param dict omitting None and empty-string values.""" + return {k: str(v) for k, v in kwargs.items() if v is not None and v != ""} # --------------------------------------------------------------------------- @@ -28,14 +38,19 @@ def _build_params(**kwargs: Any) -> dict[str, str]: async def github_list_rulesets( repo_owner: str, repo_name: str, + per_page: int | None = None, + page: int | None = None, ) -> list[dict[str, Any]] | str: """List rulesets defined on a repository.""" logger.debug("Getting rulesets for %s/%s", repo_owner, repo_name) + params = _build_params(per_page=per_page, page=page) + try: async with github_client_context() as client: response = await client.get( - f"/repos/{repo_owner}/{repo_name}/rulesets" + f"/repos/{repo_owner}/{repo_name}/rulesets", + params=params if params else None, ) if response.status == 200: @@ -143,6 +158,8 @@ async def github_list_code_scanning_alerts( severity: str | None = None, tool_name: str | None = None, ref: str | None = None, + per_page: int | None = None, + page: int | None = None, ) -> list[dict[str, Any]] | str: """List code scanning alerts with optional filters. @@ -151,13 +168,16 @@ async def github_list_code_scanning_alerts( severity: Filter by severity ('critical', 'high', 'medium', 'low', 'warning', 'note', 'error'). tool_name: Filter by code-scanning tool name (e.g. 'CodeQL'). ref: Filter by Git ref (branch name or refs/pull/N/head). + per_page: Results per page (max 100). + page: Page number (1-based). """ logger.debug( "Listing code scanning alerts for %s/%s", repo_owner, repo_name ) params = _build_params( - state=state, severity=severity, tool_name=tool_name, ref=ref + state=state, severity=severity, tool_name=tool_name, ref=ref, + per_page=per_page, page=page, ) try: @@ -203,18 +223,22 @@ async def github_list_code_scanning_analyses( repo_name: str, ref: str | None = None, tool_name: str | None = None, + per_page: int | None = None, + page: int | None = None, ) -> list[dict[str, Any]] | str: """List code scanning analyses for a repository. Args: ref: Git ref to filter by (branch name or refs/pull/N/head). tool_name: Code-scanning tool name to filter by (e.g. 'CodeQL'). + per_page: Results per page (max 100). + page: Page number (1-based). """ logger.debug( "Listing code scanning analyses for %s/%s", repo_owner, repo_name ) - params = _build_params(ref=ref, tool_name=tool_name) + params = _build_params(ref=ref, tool_name=tool_name, per_page=per_page, page=page) try: async with github_client_context() as client: @@ -311,17 +335,21 @@ async def github_list_secret_scanning_alerts( repo_owner: str, repo_name: str, state: str | None = None, + per_page: int | None = None, + page: int | None = None, ) -> list[dict[str, Any]] | str: """List secret scanning alerts for a repository. Args: state: Filter by alert state ('open' or 'resolved'). + per_page: Results per page (max 100). + page: Page number (1-based). """ logger.debug( "Listing secret scanning alerts for %s/%s", repo_owner, repo_name ) - params = _build_params(state=state) + params = _build_params(state=state, per_page=per_page, page=page) try: async with github_client_context() as client: diff --git a/tests/unit/github/test_github_ghas_forensics.py b/tests/unit/github/test_github_ghas_forensics.py index 054b2878..2bf354e3 100644 --- a/tests/unit/github/test_github_ghas_forensics.py +++ b/tests/unit/github/test_github_ghas_forensics.py @@ -64,7 +64,8 @@ async def test_list_rulesets_returns_list_when_status_200(self): assert result == rulesets mock_client.get.assert_called_once_with( - f"/repos/{_OWNER}/{_REPO}/rulesets" + f"/repos/{_OWNER}/{_REPO}/rulesets", + params=None, ) @pytest.mark.asyncio @@ -83,6 +84,64 @@ async def test_list_rulesets_returns_not_found_when_status_404(self): assert "not found" in result.lower() assert f"{_OWNER}/{_REPO}" in result + @pytest.mark.asyncio + async def test_list_rulesets_forwards_pagination_params(self): + """per_page and page are forwarded as query params when set.""" + mock_client, _ = _mock_client("get", 200, json_body=[]) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + await github_list_rulesets( + repo_owner=_OWNER, repo_name=_REPO, per_page=50, page=2 + ) + + params = mock_client.get.call_args.kwargs["params"] + assert params["per_page"] == "50" + assert params["page"] == "2" + + @pytest.mark.asyncio + async def test_list_rulesets_omits_none_pagination(self): + """None per_page/page must not appear in params.""" + mock_client, _ = _mock_client("get", 200, json_body=[]) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + await github_list_rulesets( + repo_owner=_OWNER, repo_name=_REPO, per_page=None, page=None + ) + + params = mock_client.get.call_args.kwargs["params"] + assert params is None + + @pytest.mark.asyncio + async def test_list_rulesets_returns_error_when_status_403(self): + """GET 403 (insufficient security_events scope) returns an error string.""" + mock_client, _ = _mock_client("get", 403, text_body="Forbidden") + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_list_rulesets(repo_owner=_OWNER, repo_name=_REPO) + + assert "❌" in result + assert "403" in result + + @pytest.mark.asyncio + async def test_list_rulesets_returns_error_on_connection_exception(self): + """A ConnectionError raised by the client returns a network-error string.""" + mock_client = MagicMock() + mock_client.get = AsyncMock(side_effect=ConnectionError("timeout")) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_list_rulesets(repo_owner=_OWNER, repo_name=_REPO) + + assert "❌" in result + assert "Network connection failed" in result + # =========================================================================== # github_get_ruleset @@ -213,8 +272,7 @@ async def test_list_alerts_passes_filters_as_query_params(self): ref="refs/pull/207/head", ) - call_kwargs = mock_client.get.call_args - params = call_kwargs[1]["params"] + params = mock_client.get.call_args.kwargs["params"] assert params["state"] == "open" assert params["severity"] == "critical" assert params["tool_name"] == "CodeQL" @@ -237,8 +295,7 @@ async def test_list_alerts_omits_none_filters(self): ref=None, ) - call_kwargs = mock_client.get.call_args - params = call_kwargs[1]["params"] + params = mock_client.get.call_args.kwargs["params"] assert "severity" not in params assert "tool_name" not in params assert "ref" not in params @@ -258,6 +315,53 @@ async def test_list_alerts_returns_not_found_when_status_404(self): assert "❌" in result assert f"{_OWNER}/{_REPO}" in result + @pytest.mark.asyncio + async def test_list_alerts_forwards_pagination_params(self): + """per_page and page are forwarded as query params when set.""" + mock_client, _ = _mock_client("get", 200, json_body=[]) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + await github_list_code_scanning_alerts( + repo_owner=_OWNER, repo_name=_REPO, per_page=100, page=3 + ) + + params = mock_client.get.call_args.kwargs["params"] + assert params["per_page"] == "100" + assert params["page"] == "3" + + @pytest.mark.asyncio + async def test_list_alerts_returns_error_when_status_403(self): + """GET 403 (insufficient security_events scope) returns an error string.""" + mock_client, _ = _mock_client("get", 403, text_body="Forbidden") + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_list_code_scanning_alerts( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert "❌" in result + assert "403" in result + + @pytest.mark.asyncio + async def test_list_alerts_returns_error_on_exception(self): + """An unexpected exception raised by the client returns an error string.""" + mock_client = MagicMock() + mock_client.get = AsyncMock(side_effect=RuntimeError("unexpected")) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_list_code_scanning_alerts( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert "❌" in result + assert "Error listing code scanning alerts" in result + # =========================================================================== # github_list_code_scanning_analyses @@ -301,8 +405,7 @@ async def test_list_analyses_passes_filters_as_query_params(self): tool_name="CodeQL", ) - call_kwargs = mock_client.get.call_args - params = call_kwargs[1]["params"] + params = mock_client.get.call_args.kwargs["params"] assert params["ref"] == "refs/pull/207/head" assert params["tool_name"] == "CodeQL" @@ -321,8 +424,7 @@ async def test_list_analyses_omits_none_filters(self): tool_name=None, ) - call_kwargs = mock_client.get.call_args - params = call_kwargs[1]["params"] + params = mock_client.get.call_args.kwargs["params"] assert params is None @pytest.mark.asyncio @@ -340,6 +442,22 @@ async def test_list_analyses_returns_not_found_when_status_404(self): assert "❌" in result assert f"{_OWNER}/{_REPO}" in result + @pytest.mark.asyncio + async def test_list_analyses_forwards_pagination_params(self): + """per_page and page are forwarded as query params when set.""" + mock_client, _ = _mock_client("get", 200, json_body=[]) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + await github_list_code_scanning_analyses( + repo_owner=_OWNER, repo_name=_REPO, per_page=30, page=2 + ) + + params = mock_client.get.call_args.kwargs["params"] + assert params["per_page"] == "30" + assert params["page"] == "2" + # =========================================================================== # github_get_code_scanning_default_setup @@ -422,8 +540,7 @@ async def test_list_secret_alerts_passes_state_filter(self): repo_owner=_OWNER, repo_name=_REPO, state="open" ) - call_kwargs = mock_client.get.call_args - params = call_kwargs[1]["params"] + params = mock_client.get.call_args.kwargs["params"] assert params["state"] == "open" @pytest.mark.asyncio @@ -438,8 +555,7 @@ async def test_list_secret_alerts_omits_none_state(self): repo_owner=_OWNER, repo_name=_REPO, state=None ) - call_kwargs = mock_client.get.call_args - params = call_kwargs[1]["params"] + params = mock_client.get.call_args.kwargs["params"] assert params is None @pytest.mark.asyncio @@ -456,3 +572,34 @@ async def test_list_secret_alerts_returns_not_found_when_status_404(self): assert "❌" in result assert f"{_OWNER}/{_REPO}" in result + + @pytest.mark.asyncio + async def test_list_secret_alerts_forwards_pagination_params(self): + """per_page and page are forwarded as query params when set.""" + mock_client, _ = _mock_client("get", 200, json_body=[]) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + await github_list_secret_scanning_alerts( + repo_owner=_OWNER, repo_name=_REPO, per_page=100, page=1 + ) + + params = mock_client.get.call_args.kwargs["params"] + assert params["per_page"] == "100" + assert params["page"] == "1" + + @pytest.mark.asyncio + async def test_list_secret_alerts_returns_error_when_status_403(self): + """GET 403 (insufficient security_events scope) returns an error string.""" + mock_client, _ = _mock_client("get", 403, text_body="Forbidden") + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_list_secret_scanning_alerts( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert "❌" in result + assert "403" in result From b349cc2972a852fb9ec2a838a7f4780cb139a591 Mon Sep 17 00:00:00 2001 From: mementorc Date: Fri, 19 Jun 2026 11:22:28 -0500 Subject: [PATCH 3/3] test(github): complete 403/exception coverage for GHAS forensics tools Adds 403 tests for github_get_ruleset, github_get_branch_rules, github_list_code_scanning_analyses, github_get_code_scanning_default_setup, and an exception-path test for default_setup, so every GHAS tool has uniform error-path coverage (PR #188 follow-up). Co-Authored-By: Claude Opus 4.8 --- .../unit/github/test_github_ghas_forensics.py | 76 +++++++++++++++++++ 1 file changed, 76 insertions(+) diff --git a/tests/unit/github/test_github_ghas_forensics.py b/tests/unit/github/test_github_ghas_forensics.py index 2bf354e3..d7be6edd 100644 --- a/tests/unit/github/test_github_ghas_forensics.py +++ b/tests/unit/github/test_github_ghas_forensics.py @@ -185,6 +185,21 @@ async def test_get_ruleset_returns_not_found_when_status_404(self): assert "42" in result assert f"{_OWNER}/{_REPO}" in result + @pytest.mark.asyncio + async def test_get_ruleset_returns_error_when_status_403(self): + """GET 403 (insufficient security_events scope) returns an error string.""" + mock_client, _ = _mock_client("get", 403, text_body="Forbidden") + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_get_ruleset( + repo_owner=_OWNER, repo_name=_REPO, ruleset_id=42 + ) + + assert "❌" in result + assert "403" in result + # =========================================================================== # github_get_branch_rules @@ -227,6 +242,21 @@ async def test_get_branch_rules_returns_not_found_when_status_404(self): assert "❌" in result assert f"{_OWNER}/{_REPO}#main" in result + @pytest.mark.asyncio + async def test_get_branch_rules_returns_error_when_status_403(self): + """GET 403 (insufficient security_events scope) returns an error string.""" + mock_client, _ = _mock_client("get", 403, text_body="Forbidden") + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_get_branch_rules( + repo_owner=_OWNER, repo_name=_REPO, branch="main" + ) + + assert "❌" in result + assert "403" in result + # =========================================================================== # github_list_code_scanning_alerts @@ -458,6 +488,21 @@ async def test_list_analyses_forwards_pagination_params(self): assert params["per_page"] == "30" assert params["page"] == "2" + @pytest.mark.asyncio + async def test_list_analyses_returns_error_when_status_403(self): + """GET 403 (insufficient security_events scope) returns an error string.""" + mock_client, _ = _mock_client("get", 403, text_body="Forbidden") + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_list_code_scanning_analyses( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert "❌" in result + assert "403" in result + # =========================================================================== # github_get_code_scanning_default_setup @@ -500,6 +545,37 @@ async def test_get_default_setup_returns_not_found_when_status_404(self): assert "❌" in result assert f"{_OWNER}/{_REPO}" in result + @pytest.mark.asyncio + async def test_get_default_setup_returns_error_when_status_403(self): + """GET 403 (insufficient security_events scope) returns an error string.""" + mock_client, _ = _mock_client("get", 403, text_body="Forbidden") + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_get_code_scanning_default_setup( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert "❌" in result + assert "403" in result + + @pytest.mark.asyncio + async def test_get_default_setup_returns_error_on_exception(self): + """An unexpected exception raised by the client returns an error string.""" + mock_client = MagicMock() + mock_client.get = AsyncMock(side_effect=ConnectionError("refused")) + + with patch(_PATCH_CTX) as mock_ctx: + mock_ctx.return_value.__aenter__.return_value = mock_client + + result = await github_get_code_scanning_default_setup( + repo_owner=_OWNER, repo_name=_REPO + ) + + assert "❌" in result + assert "Network connection failed" in result + # =========================================================================== # github_list_secret_scanning_alerts