diff --git a/env.example b/env.example index c7a1517..b9b6bb9 100644 --- a/env.example +++ b/env.example @@ -1,8 +1,15 @@ # Note: If you have these environment variables set in your own environment, the ones in your environment will take precedence over any you put here. # GitHub Configuration +# Option 1: Single token (simplest) GITHUB_TOKEN=your_github_token_here +# Option 2: Multiple tokens (for working across multiple projects with fine-grained PATs) +# Comma-separated list of tokens. When --post is used, each token is tested for write access +# and the first one that works is used. This is useful if you have separate fine-grained PATs +# for different repositories. Takes precedence over GITHUB_TOKEN for write operations. +# REVIEW_ROADMAP_GITHUB_TOKENS=ghp_token1,ghp_token2,ghp_token3 + # Application Settings REVIEW_ROADMAP_LOG_LEVEL=INFO diff --git a/review_roadmap/config.py b/review_roadmap/config.py index b5757d6..87f894c 100644 --- a/review_roadmap/config.py +++ b/review_roadmap/config.py @@ -7,7 +7,8 @@ import os from pathlib import Path -from typing import Optional +from typing import List, Optional +from pydantic import model_validator from pydantic_settings import BaseSettings, SettingsConfigDict @@ -21,6 +22,9 @@ class Settings(BaseSettings): Attributes: GITHUB_TOKEN: GitHub API token for fetching PR data. + REVIEW_ROADMAP_GITHUB_TOKENS: Comma-separated list of GitHub tokens. + When set, these are tried in order during write access checks. + Takes precedence over GITHUB_TOKEN for write operations. REVIEW_ROADMAP_LLM_PROVIDER: LLM provider to use. Options: 'anthropic', 'anthropic-vertex', 'openai', 'google'. REVIEW_ROADMAP_MODEL_NAME: Model name (e.g., 'claude-opus-4-5', 'gpt-4o'). @@ -39,7 +43,60 @@ class Settings(BaseSettings): ) # GitHub - GITHUB_TOKEN: str + GITHUB_TOKEN: Optional[str] = None + REVIEW_ROADMAP_GITHUB_TOKENS: Optional[str] = None + + @model_validator(mode="after") + def validate_github_token(self) -> "Settings": + """Ensure at least one GitHub token is configured.""" + if not self.GITHUB_TOKEN and not self.REVIEW_ROADMAP_GITHUB_TOKENS: + raise ValueError( + "Either GITHUB_TOKEN or REVIEW_ROADMAP_GITHUB_TOKENS must be set" + ) + return self + + def get_github_tokens(self) -> List[str]: + """Get the list of GitHub tokens to try, in order of precedence. + + When REVIEW_ROADMAP_GITHUB_TOKENS is set, those tokens take precedence + and are returned first. GITHUB_TOKEN (if set and not already in the list) + is appended as a fallback. + + Returns: + List of unique, non-empty GitHub tokens to try. + """ + tokens: List[str] = [] + + # REVIEW_ROADMAP_GITHUB_TOKENS takes precedence + if self.REVIEW_ROADMAP_GITHUB_TOKENS: + tokens.extend( + t.strip() + for t in self.REVIEW_ROADMAP_GITHUB_TOKENS.split(",") + if t.strip() + ) + + # Add GITHUB_TOKEN as fallback if not already included + if self.GITHUB_TOKEN and self.GITHUB_TOKEN not in tokens: + tokens.append(self.GITHUB_TOKEN) + + return tokens + + def get_default_github_token(self) -> str: + """Get the default GitHub token for read operations. + + Returns the first available token (REVIEW_ROADMAP_GITHUB_TOKENS + takes precedence over GITHUB_TOKEN). + + Returns: + The first available GitHub token. + + Raises: + ValueError: If no tokens are configured. + """ + tokens = self.get_github_tokens() + if not tokens: + raise ValueError("No GitHub tokens configured") + return tokens[0] # LLM Configuration (prefixed to avoid conflicts with shell environment) REVIEW_ROADMAP_LLM_PROVIDER: str = "anthropic" diff --git a/review_roadmap/github/client.py b/review_roadmap/github/client.py index aba6090..645f066 100644 --- a/review_roadmap/github/client.py +++ b/review_roadmap/github/client.py @@ -5,6 +5,7 @@ generate review roadmaps. """ +from dataclasses import dataclass from typing import Any, Dict, List, Optional, Tuple import httpx @@ -16,6 +17,21 @@ ) +@dataclass +class TokenSearchResult: + """Result of searching for a token with write access. + + Attributes: + token: The token that was found with write access, or None if no token worked. + access_result: The WriteAccessResult from checking the successful token, + or the last failed result if no token worked. + tokens_tried: Number of tokens that were tested. + """ + token: Optional[str] + access_result: WriteAccessResult + tokens_tried: int + + class GitHubClient: """Synchronous GitHub API client for PR data retrieval. @@ -37,9 +53,10 @@ def __init__(self, token: Optional[str] = None): """Initialize the GitHub client. Args: - token: GitHub API token. If not provided, uses GITHUB_TOKEN from settings. + token: GitHub API token. If not provided, uses the first available + token from settings (REVIEW_ROADMAP_GITHUB_TOKENS takes precedence). """ - self.token = token or settings.GITHUB_TOKEN + self.token = token or settings.get_default_github_token() self.headers = { "Authorization": f"Bearer {self.token}", "Accept": "application/vnd.github.v3+json", @@ -471,3 +488,79 @@ def post_pr_comment(self, owner: str, repo: str, pr_number: int, body: str) -> D ) resp.raise_for_status() return resp.json() + + +def find_working_token( + owner: str, repo: str, pr_number: int +) -> TokenSearchResult: + """Find a GitHub token with write access from the configured tokens. + + Iterates through all configured tokens (from REVIEW_ROADMAP_GITHUB_TOKENS + and GITHUB_TOKEN) and tests each one for write access using the + check_write_access method with a live reaction test. + + Args: + owner: Repository owner. + repo: Repository name. + pr_number: PR number for live write testing. + + Returns: + TokenSearchResult containing: + - token: The first token with GRANTED status, or None if none worked + - access_result: The WriteAccessResult from the successful check, + or the last failed result if no token worked + - tokens_tried: Number of tokens that were tested + + Example: + >>> result = find_working_token("owner", "repo", 123) + >>> if result.token: + ... client = GitHubClient(token=result.token) + ... client.post_pr_comment(owner, repo, 123, "Hello!") + """ + tokens = settings.get_github_tokens() + + if not tokens: + return TokenSearchResult( + token=None, + access_result=WriteAccessResult( + status=WriteAccessStatus.DENIED, + is_fine_grained_pat=False, + message="No GitHub tokens configured." + ), + tokens_tried=0 + ) + + last_result: Optional[WriteAccessResult] = None + + for i, token in enumerate(tokens, start=1): + client = GitHubClient(token=token) + try: + result = client.check_write_access(owner, repo, pr_number) + last_result = result + + if result.status == WriteAccessStatus.GRANTED: + return TokenSearchResult( + token=token, + access_result=result, + tokens_tried=i + ) + except Exception: + # Token failed to even check access (e.g., network error, invalid token) + # Continue to next token + last_result = WriteAccessResult( + status=WriteAccessStatus.DENIED, + is_fine_grained_pat=False, + message=f"Token failed basic validation (may be invalid or revoked)." + ) + continue + + # No token worked - return the last result + return TokenSearchResult( + token=None, + access_result=last_result or WriteAccessResult( + status=WriteAccessStatus.DENIED, + is_fine_grained_pat=False, + message="No tokens were successfully tested." + ), + tokens_tried=len(tokens) + ) diff --git a/review_roadmap/main.py b/review_roadmap/main.py index 373420c..70d99c7 100644 --- a/review_roadmap/main.py +++ b/review_roadmap/main.py @@ -1,7 +1,7 @@ import typer from rich.console import Console from rich.markdown import Markdown -from review_roadmap.github.client import GitHubClient +from review_roadmap.github.client import GitHubClient, find_working_token from review_roadmap.agent.graph import build_graph from review_roadmap.config import settings from review_roadmap.logging import configure_logging @@ -59,27 +59,49 @@ def generate( console.print("[red]Invalid PR format. Use 'owner/repo/number' or a full URL.[/red]") raise typer.Exit(code=1) - # Initialize GitHub client + # Initialize GitHub client (default token for read operations) gh_client = GitHubClient() + + # Token with write access (may differ from default if using multi-token) + write_token = None # Check write access early if posting is requested (fail fast before LLM generation) if post: - console.print(f"[bold blue]Checking write access for {owner}/{repo}...[/bold blue]") + tokens = settings.get_github_tokens() + + if len(tokens) > 1: + console.print(f"[bold blue]Searching {len(tokens)} tokens for write access to {owner}/{repo}...[/bold blue]") + else: + console.print(f"[bold blue]Checking write access for {owner}/{repo}...[/bold blue]") + try: - # Pass pr_number to enable live write test for fine-grained PATs - access_result = gh_client.check_write_access(owner, repo, pr_number) + # Search through available tokens for one with write access + search_result = find_working_token(owner, repo, pr_number) - if access_result.status == WriteAccessStatus.DENIED: - console.print( - f"[red]Error: {access_result.message}[/red]\n" - "[yellow]To use --post, your token needs 'Pull requests: Read and write' permission.[/yellow]" - ) - raise typer.Exit(code=1) - elif access_result.status == WriteAccessStatus.UNCERTAIN: - console.print(f"[yellow]Warning: {access_result.message}[/yellow]") + if search_result.token: + write_token = search_result.token + if search_result.tokens_tried > 1: + console.print(f"[green]Write access confirmed (token {search_result.tokens_tried} of {len(tokens)}).[/green]") + else: + console.print("[green]Write access confirmed.[/green]") + # Update client to use the working token for subsequent operations + gh_client = GitHubClient(token=write_token) + elif search_result.access_result.status == WriteAccessStatus.UNCERTAIN: + console.print(f"[yellow]Warning: {search_result.access_result.message}[/yellow]") console.print("[yellow]Proceeding, but posting may fail...[/yellow]") else: - console.print("[green]Write access confirmed.[/green]") + if search_result.tokens_tried > 1: + console.print( + f"[red]Error: None of the {search_result.tokens_tried} configured tokens have write access.[/red]\n" + f"[red]Last error: {search_result.access_result.message}[/red]\n" + "[yellow]To use --post, at least one token needs 'Pull requests: Read and write' permission.[/yellow]" + ) + else: + console.print( + f"[red]Error: {search_result.access_result.message}[/red]\n" + "[yellow]To use --post, your token needs 'Pull requests: Read and write' permission.[/yellow]" + ) + raise typer.Exit(code=1) except typer.Exit: raise except Exception as e: diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..89e7a73 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,143 @@ +"""Tests for the config module.""" + +import os +import pytest +from review_roadmap.config import Settings + + +class TestGetGithubTokens: + """Tests for Settings.get_github_tokens method.""" + + def test_get_github_tokens_single_token(self, monkeypatch): + """Returns single GITHUB_TOKEN when no multi-token configured.""" + # Clear any env vars that might interfere + monkeypatch.delenv("GITHUB_TOKEN", raising=False) + monkeypatch.delenv("REVIEW_ROADMAP_GITHUB_TOKENS", raising=False) + + settings = Settings( + GITHUB_TOKEN="single-token", + REVIEW_ROADMAP_MODEL_NAME="test-model", + _env_file=None # Disable .env file loading + ) + + tokens = settings.get_github_tokens() + + assert tokens == ["single-token"] + + def test_get_github_tokens_multi_token_precedence(self, monkeypatch): + """REVIEW_ROADMAP_GITHUB_TOKENS takes precedence over GITHUB_TOKEN.""" + monkeypatch.delenv("GITHUB_TOKEN", raising=False) + monkeypatch.delenv("REVIEW_ROADMAP_GITHUB_TOKENS", raising=False) + + settings = Settings( + GITHUB_TOKEN="fallback-token", + REVIEW_ROADMAP_GITHUB_TOKENS="token1,token2,token3", + REVIEW_ROADMAP_MODEL_NAME="test-model", + _env_file=None + ) + + tokens = settings.get_github_tokens() + + # Multi-tokens should come first, fallback token appended + assert tokens == ["token1", "token2", "token3", "fallback-token"] + + def test_get_github_tokens_strips_whitespace(self, monkeypatch): + """Whitespace is stripped from comma-separated tokens.""" + monkeypatch.delenv("GITHUB_TOKEN", raising=False) + monkeypatch.delenv("REVIEW_ROADMAP_GITHUB_TOKENS", raising=False) + + settings = Settings( + REVIEW_ROADMAP_GITHUB_TOKENS=" token1 , token2 , token3 ", + REVIEW_ROADMAP_MODEL_NAME="test-model", + _env_file=None + ) + + tokens = settings.get_github_tokens() + + assert tokens == ["token1", "token2", "token3"] + + def test_get_github_tokens_no_duplicates(self, monkeypatch): + """GITHUB_TOKEN is not duplicated if already in REVIEW_ROADMAP_GITHUB_TOKENS.""" + monkeypatch.delenv("GITHUB_TOKEN", raising=False) + monkeypatch.delenv("REVIEW_ROADMAP_GITHUB_TOKENS", raising=False) + + settings = Settings( + GITHUB_TOKEN="token1", + REVIEW_ROADMAP_GITHUB_TOKENS="token1,token2", + REVIEW_ROADMAP_MODEL_NAME="test-model", + _env_file=None + ) + + tokens = settings.get_github_tokens() + + # token1 should not be duplicated + assert tokens == ["token1", "token2"] + + def test_get_github_tokens_skips_empty(self, monkeypatch): + """Empty tokens from extra commas are skipped.""" + monkeypatch.delenv("GITHUB_TOKEN", raising=False) + monkeypatch.delenv("REVIEW_ROADMAP_GITHUB_TOKENS", raising=False) + + settings = Settings( + REVIEW_ROADMAP_GITHUB_TOKENS="token1,,token2,", + REVIEW_ROADMAP_MODEL_NAME="test-model", + _env_file=None + ) + + tokens = settings.get_github_tokens() + + assert tokens == ["token1", "token2"] + + +class TestGetDefaultGithubToken: + """Tests for Settings.get_default_github_token method.""" + + def test_get_default_github_token_returns_first(self, monkeypatch): + """Returns the first token from the tokens list.""" + monkeypatch.delenv("GITHUB_TOKEN", raising=False) + monkeypatch.delenv("REVIEW_ROADMAP_GITHUB_TOKENS", raising=False) + + settings = Settings( + REVIEW_ROADMAP_GITHUB_TOKENS="first-token,second-token", + REVIEW_ROADMAP_MODEL_NAME="test-model", + _env_file=None + ) + + token = settings.get_default_github_token() + + assert token == "first-token" + + def test_get_default_github_token_uses_github_token_fallback(self, monkeypatch): + """Falls back to GITHUB_TOKEN when no multi-token configured.""" + monkeypatch.delenv("GITHUB_TOKEN", raising=False) + monkeypatch.delenv("REVIEW_ROADMAP_GITHUB_TOKENS", raising=False) + + settings = Settings( + GITHUB_TOKEN="my-github-token", + REVIEW_ROADMAP_MODEL_NAME="test-model", + _env_file=None + ) + + token = settings.get_default_github_token() + + assert token == "my-github-token" + + +class TestSettingsValidation: + """Tests for Settings validation.""" + + def test_requires_at_least_one_token(self, monkeypatch): + """Raises error when no GitHub tokens are configured.""" + # Must clear env vars so pydantic-settings doesn't pick them up + monkeypatch.delenv("GITHUB_TOKEN", raising=False) + monkeypatch.delenv("REVIEW_ROADMAP_GITHUB_TOKENS", raising=False) + + with pytest.raises(ValueError) as exc_info: + Settings( + REVIEW_ROADMAP_MODEL_NAME="test-model", + _env_file=None + # No GITHUB_TOKEN or REVIEW_ROADMAP_GITHUB_TOKENS + ) + + # Should mention needing a token + assert "GITHUB_TOKEN" in str(exc_info.value) or "REVIEW_ROADMAP_GITHUB_TOKENS" in str(exc_info.value) diff --git a/tests/test_github_client.py b/tests/test_github_client.py index b6c1c78..6c4448e 100644 --- a/tests/test_github_client.py +++ b/tests/test_github_client.py @@ -623,3 +623,158 @@ def test_minimize_old_roadmap_comments_fetch_error(): assert minimized == 0 assert errors == 0 + + +# --- Tests for find_working_token --- + +from unittest.mock import patch +from review_roadmap.github.client import find_working_token, TokenSearchResult +from review_roadmap.models import WriteAccessResult + + +@respx.mock +def test_find_working_token_first_token_works(): + """find_working_token returns first token when it has write access.""" + owner = "owner" + repo = "repo" + pr_number = 42 + + # Mock repo endpoint with push permission + respx.get(f"https://api.github.com/repos/{owner}/{repo}").mock( + return_value=Response(200, json={ + "id": 12345, + "name": repo, + "private": False, + "permissions": {"admin": False, "push": True, "pull": True} + }, headers={"X-OAuth-Scopes": "repo"}) + ) + + with patch("review_roadmap.github.client.settings") as mock_settings: + mock_settings.get_github_tokens.return_value = ["token1", "token2"] + + result = find_working_token(owner, repo, pr_number) + + assert result.token == "token1" + assert result.access_result.status == WriteAccessStatus.GRANTED + assert result.tokens_tried == 1 + + +@respx.mock +def test_find_working_token_second_token_works(): + """find_working_token tries second token when first fails.""" + owner = "owner" + repo = "repo" + pr_number = 42 + + # Track which token is being used via call count + call_count = [0] + + def repo_response(request): + call_count[0] += 1 + if call_count[0] == 1: + # First token: no push permission + return Response(200, json={ + "id": 12345, + "name": repo, + "private": False, + "permissions": {"admin": False, "push": False, "pull": True} + }) + else: + # Second token: has push permission + return Response(200, json={ + "id": 12345, + "name": repo, + "private": False, + "permissions": {"admin": False, "push": True, "pull": True} + }, headers={"X-OAuth-Scopes": "repo"}) + + respx.get(f"https://api.github.com/repos/{owner}/{repo}").mock(side_effect=repo_response) + + with patch("review_roadmap.github.client.settings") as mock_settings: + mock_settings.get_github_tokens.return_value = ["token1", "token2"] + + result = find_working_token(owner, repo, pr_number) + + assert result.token == "token2" + assert result.access_result.status == WriteAccessStatus.GRANTED + assert result.tokens_tried == 2 + + +@respx.mock +def test_find_working_token_no_tokens_configured(): + """find_working_token returns None when no tokens configured.""" + owner = "owner" + repo = "repo" + pr_number = 42 + + with patch("review_roadmap.github.client.settings") as mock_settings: + mock_settings.get_github_tokens.return_value = [] + + result = find_working_token(owner, repo, pr_number) + + assert result.token is None + assert result.access_result.status == WriteAccessStatus.DENIED + assert result.tokens_tried == 0 + assert "No GitHub tokens configured" in result.access_result.message + + +@respx.mock +def test_find_working_token_all_tokens_fail(): + """find_working_token returns None when all tokens fail.""" + owner = "owner" + repo = "repo" + pr_number = 42 + + # All tokens have no push permission + respx.get(f"https://api.github.com/repos/{owner}/{repo}").mock( + return_value=Response(200, json={ + "id": 12345, + "name": repo, + "private": False, + "permissions": {"admin": False, "push": False, "pull": True} + }) + ) + + with patch("review_roadmap.github.client.settings") as mock_settings: + mock_settings.get_github_tokens.return_value = ["token1", "token2", "token3"] + + result = find_working_token(owner, repo, pr_number) + + assert result.token is None + assert result.access_result.status == WriteAccessStatus.DENIED + assert result.tokens_tried == 3 + + +@respx.mock +def test_find_working_token_handles_invalid_token(): + """find_working_token handles invalid/revoked tokens gracefully.""" + owner = "owner" + repo = "repo" + pr_number = 42 + + call_count = [0] + + def repo_response(request): + call_count[0] += 1 + if call_count[0] == 1: + # First token is invalid + return Response(401, json={"message": "Bad credentials"}) + else: + # Second token works + return Response(200, json={ + "id": 12345, + "name": repo, + "private": False, + "permissions": {"admin": False, "push": True, "pull": True} + }, headers={"X-OAuth-Scopes": "repo"}) + + respx.get(f"https://api.github.com/repos/{owner}/{repo}").mock(side_effect=repo_response) + + with patch("review_roadmap.github.client.settings") as mock_settings: + mock_settings.get_github_tokens.return_value = ["bad-token", "good-token"] + + result = find_working_token(owner, repo, pr_number) + + assert result.token == "good-token" + assert result.access_result.status == WriteAccessStatus.GRANTED + assert result.tokens_tried == 2 diff --git a/tests/test_main.py b/tests/test_main.py index ad16d03..0ef4d20 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -198,52 +198,70 @@ def test_generate_posts_to_pr(self): with patch("review_roadmap.main.GitHubClient") as mock_gh: mock_client = MagicMock() mock_client.get_pr_context.return_value = mock_context - from review_roadmap.models import WriteAccessResult, WriteAccessStatus - mock_client.check_write_access.return_value = WriteAccessResult( - status=WriteAccessStatus.GRANTED, - is_fine_grained_pat=False, - message="Classic token with correct scopes verified." - ) mock_gh.return_value = mock_client - with patch("review_roadmap.main.build_graph") as mock_build: - mock_graph = MagicMock() - mock_graph.invoke.return_value = mock_graph_result - mock_build.return_value = mock_graph + with patch("review_roadmap.main.find_working_token") as mock_find_token: + from review_roadmap.models import WriteAccessResult, WriteAccessStatus + from review_roadmap.github.client import TokenSearchResult + mock_find_token.return_value = TokenSearchResult( + token="test-token", + access_result=WriteAccessResult( + status=WriteAccessStatus.GRANTED, + is_fine_grained_pat=False, + message="Classic token with correct scopes verified." + ), + tokens_tried=1 + ) - with patch("review_roadmap.main.settings") as mock_settings: - mock_settings.REVIEW_ROADMAP_LLM_PROVIDER = "anthropic" - mock_settings.REVIEW_ROADMAP_MODEL_NAME = "claude" + with patch("review_roadmap.main.build_graph") as mock_build: + mock_graph = MagicMock() + mock_graph.invoke.return_value = mock_graph_result + mock_build.return_value = mock_graph - from review_roadmap.main import app + with patch("review_roadmap.main.settings") as mock_settings: + mock_settings.REVIEW_ROADMAP_LLM_PROVIDER = "anthropic" + mock_settings.REVIEW_ROADMAP_MODEL_NAME = "claude" + mock_settings.get_github_tokens.return_value = ["test-token"] - result = runner.invoke(app, ["owner/repo/1", "--post"]) + from review_roadmap.main import app + + result = runner.invoke(app, ["owner/repo/1", "--post"]) - assert result.exit_code == 0 - mock_client.check_write_access.assert_called_once() - mock_client.post_pr_comment.assert_called_once() + assert result.exit_code == 0 + mock_find_token.assert_called_once() + mock_client.post_pr_comment.assert_called_once() def test_generate_post_fails_without_write_access(self): """Test that --post fails early if no write access.""" with patch("review_roadmap.main.configure_logging"): with patch("review_roadmap.main.GitHubClient") as mock_gh: mock_client = MagicMock() - from review_roadmap.models import WriteAccessResult, WriteAccessStatus - mock_client.check_write_access.return_value = WriteAccessResult( - status=WriteAccessStatus.DENIED, - is_fine_grained_pat=False, - message="Token lacks required scope." - ) mock_gh.return_value = mock_client - from review_roadmap.main import app + with patch("review_roadmap.main.find_working_token") as mock_find_token: + from review_roadmap.models import WriteAccessResult, WriteAccessStatus + from review_roadmap.github.client import TokenSearchResult + mock_find_token.return_value = TokenSearchResult( + token=None, + access_result=WriteAccessResult( + status=WriteAccessStatus.DENIED, + is_fine_grained_pat=False, + message="Token lacks required scope." + ), + tokens_tried=1 + ) - result = runner.invoke(app, ["owner/repo/1", "--post"]) + with patch("review_roadmap.main.settings") as mock_settings: + mock_settings.get_github_tokens.return_value = ["test-token"] - assert result.exit_code == 1 - assert "write access" in result.output.lower() - # Should not have tried to fetch PR context - mock_client.get_pr_context.assert_not_called() + from review_roadmap.main import app + + result = runner.invoke(app, ["owner/repo/1", "--post"]) + + assert result.exit_code == 1 + assert "write access" in result.output.lower() + # Should not have tried to fetch PR context + mock_client.get_pr_context.assert_not_called() def test_generate_handles_pr_fetch_error(self): """Test that PR fetch errors are handled gracefully.""" @@ -272,30 +290,38 @@ def test_generate_handles_post_error(self): with patch("review_roadmap.main.GitHubClient") as mock_gh: mock_client = MagicMock() mock_client.get_pr_context.return_value = mock_context - from review_roadmap.models import WriteAccessResult, WriteAccessStatus - mock_client.check_write_access.return_value = WriteAccessResult( - status=WriteAccessStatus.GRANTED, - is_fine_grained_pat=False, - message="Classic token with correct scopes verified." - ) mock_client.post_pr_comment.side_effect = Exception("Post failed") mock_gh.return_value = mock_client - with patch("review_roadmap.main.build_graph") as mock_build: - mock_graph = MagicMock() - mock_graph.invoke.return_value = mock_graph_result - mock_build.return_value = mock_graph + with patch("review_roadmap.main.find_working_token") as mock_find_token: + from review_roadmap.models import WriteAccessResult, WriteAccessStatus + from review_roadmap.github.client import TokenSearchResult + mock_find_token.return_value = TokenSearchResult( + token="test-token", + access_result=WriteAccessResult( + status=WriteAccessStatus.GRANTED, + is_fine_grained_pat=False, + message="Classic token with correct scopes verified." + ), + tokens_tried=1 + ) - with patch("review_roadmap.main.settings") as mock_settings: - mock_settings.REVIEW_ROADMAP_LLM_PROVIDER = "anthropic" - mock_settings.REVIEW_ROADMAP_MODEL_NAME = "claude" + with patch("review_roadmap.main.build_graph") as mock_build: + mock_graph = MagicMock() + mock_graph.invoke.return_value = mock_graph_result + mock_build.return_value = mock_graph - from review_roadmap.main import app + with patch("review_roadmap.main.settings") as mock_settings: + mock_settings.REVIEW_ROADMAP_LLM_PROVIDER = "anthropic" + mock_settings.REVIEW_ROADMAP_MODEL_NAME = "claude" + mock_settings.get_github_tokens.return_value = ["test-token"] - result = runner.invoke(app, ["owner/repo/1", "--post"]) + from review_roadmap.main import app - assert result.exit_code == 1 - assert "Error posting comment" in result.output + result = runner.invoke(app, ["owner/repo/1", "--post"]) + + assert result.exit_code == 1 + assert "Error posting comment" in result.output def test_generate_prints_to_console_by_default(self): """Test that roadmap is printed to console when no output/post flags."""