feat: improve token validation with exponential backoff and add unit tests

gittensor/miner/token_mgmt.py

@@ -2,22 +2,30 @@
 import os
 import sys
 import time
-from typing import Optional
+from typing import Optional, Tuple
 
 import bittensor as bt
 import requests
 
 from gittensor.constants import BASE_GITHUB_API_URL
 
+# Token validation retry configuration
+MAX_RETRIES: int = 3
+INITIAL_BACKOFF_SECONDS: float = 2.0
+BACKOFF_MULTIPLIER: float = 2.0
+
+# Rate limit thresholds
+RATE_LIMIT_REMAINING_WARN: int = 100
+
 
 def init() -> bool:
-    """Initialize and check if GitHub token exists in environment
+    """Initialize and check if GitHub token exists in environment.
 
     Returns:
-        bool: Always returns True if token exists, otherwise exits
+        bool: Always returns True if token exists, otherwise exits.
 
     Raises:
-        SystemExit: If GITTENSOR_MINER_PAT environment variable is not set
+        SystemExit: If GITTENSOR_MINER_PAT environment variable is not set.
     """
     token = os.getenv('GITTENSOR_MINER_PAT')
     if not token:
@@ -30,11 +38,13 @@ def init() -> bool:
 
 
 def load_token() -> Optional[str]:
-    """
-    Load GitHub token from environment variable
+    """Load GitHub token from environment variable and validate it.
+
+    Reads the token from the ``GITTENSOR_MINER_PAT`` environment variable,
+    validates it against the GitHub API, and returns it if valid.
 
     Returns:
-        Optional[str]: The GitHub access token string if valid, None otherwise
+        Optional[str]: The GitHub access token string if valid, None otherwise.
     """
     bt.logging.info('Loading GitHub token from environment.')
 
@@ -45,33 +55,105 @@ def load_token() -> Optional[str]:
         return None
 
     # Test if token is still valid
-    if is_token_valid(access_token):
-        bt.logging.info('GitHub token loaded successfully and is valid.')
+    valid, message = validate_token(access_token)
+    if valid:
+        bt.logging.info(f'GitHub token loaded successfully and is valid. {message}')
         return access_token
 
-    bt.logging.error('GitHub token is invalid or expired.')
+    bt.logging.error(f'GitHub token is invalid or expired. {message}')
     return None
 
 
-def is_token_valid(token: str) -> bool:
-    """
-    Test if a GitHub token is valid by making a simple API call.
+def validate_token(token: str) -> Tuple[bool, str]:
+    """Validate a GitHub token and return status with diagnostic info.
+
+    Makes an authenticated request to the GitHub ``/user`` endpoint to verify
+    token validity. Implements exponential backoff on transient failures and
+    provides diagnostic information about rate limits.
 
     Args:
-        token (str): GitHub personal access token to validate
+        token: GitHub personal access token to validate.
 
     Returns:
-        bool: True if valid token, False otherwise
+        A tuple of ``(is_valid, message)`` where *message* contains
+        diagnostic information such as the authenticated username or
+        the reason for failure.
     """
     headers = {'Authorization': f'token {token}', 'Accept': 'application/vnd.github.v3+json'}
+    backoff = INITIAL_BACKOFF_SECONDS
 
-    for attempt in range(3):
+    for attempt in range(MAX_RETRIES):
         try:
             response = requests.get(f'{BASE_GITHUB_API_URL}/user', headers=headers, timeout=15)
-            return response.status_code == 200
-        except Exception as e:
-            bt.logging.warning(f'Error validating GitHub token (attempt {attempt + 1}/3): {e}')
-            if attempt < 2:  # Don't sleep on last attempt
-                time.sleep(3)
 
-    return False
+            if response.status_code == 200:
+                username = response.json().get('login', 'unknown')
+                _check_rate_limit(response)
+                return True, f'Authenticated as {username}'
+
+            if response.status_code == 401:
+                return False, 'Token is invalid or revoked (HTTP 401)'
+
+            if response.status_code == 403:
+                remaining = response.headers.get('X-RateLimit-Remaining', '?')
+                reset = response.headers.get('X-RateLimit-Reset')
+                if remaining == '0' and reset:
+                    reset_time = time.strftime('%H:%M:%S UTC', time.gmtime(int(reset)))
+                    return False, f'Rate limited (HTTP 403). Resets at {reset_time}'
+                return False, f'Forbidden (HTTP 403). Rate limit remaining: {remaining}'
+
+            bt.logging.warning(
+                f'Unexpected status {response.status_code} validating token (attempt {attempt + 1}/{MAX_RETRIES})'
+            )
+
+        except requests.exceptions.Timeout:
+            bt.logging.warning(f'Timeout validating GitHub token (attempt {attempt + 1}/{MAX_RETRIES})')
+
+        except requests.exceptions.ConnectionError as e:
+            bt.logging.warning(f'Connection error validating GitHub token (attempt {attempt + 1}/{MAX_RETRIES}): {e}')
+
+        except requests.exceptions.RequestException as e:
+            bt.logging.warning(f'Request error validating GitHub token (attempt {attempt + 1}/{MAX_RETRIES}): {e}')
+
+        if attempt < MAX_RETRIES - 1:
+            bt.logging.info(f'Retrying in {backoff:.1f}s...')
+            time.sleep(backoff)
+            backoff *= BACKOFF_MULTIPLIER
+
+    return False, f'Failed after {MAX_RETRIES} attempts'
+
+
+def is_token_valid(token: str) -> bool:
+    """Test if a GitHub token is valid by making a simple API call.
+
+    This is a convenience wrapper around :func:`validate_token` that
+    returns only the boolean result.
+
+    Args:
+        token: GitHub personal access token to validate.
+
+    Returns:
+        bool: True if valid token, False otherwise.
+    """
+    valid, _ = validate_token(token)
+    return valid
+
+
+def _check_rate_limit(response: requests.Response) -> None:
+    """Log a warning if the GitHub API rate limit is running low.
+
+    Args:
+        response: A successful GitHub API response whose headers
+            contain rate-limit information.
+    """
+    remaining = response.headers.get('X-RateLimit-Remaining')
+    limit = response.headers.get('X-RateLimit-Limit')
+    if remaining is not None:
+        try:
+            remaining_int = int(remaining)
+            if remaining_int < RATE_LIMIT_REMAINING_WARN:
+                bt.logging.warning(
+                    f'GitHub API rate limit running low: {remaining}/{limit} requests remaining'
+                )
+        except ValueError:
+            pass

tests/miner/test_token_mgmt.py

@@ -0,0 +1,147 @@
+# The MIT License (MIT)
+# Copyright © 2025 Entrius
+# GitTensor Miner Token Management Tests
+
+"""Unit tests for gittensor.miner.token_mgmt module."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from gittensor.miner.token_mgmt import (
+    _check_rate_limit,
+    validate_token,
+    is_token_valid,
+)
+
+
+class TestValidateToken:
+    """Tests for the validate_token function."""
+
+    @patch('gittensor.miner.token_mgmt.requests.get')
+    def test_valid_token(self, mock_get):
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {'login': 'testuser'}
+        mock_response.headers = {
+            'X-RateLimit-Remaining': '4999',
+            'X-RateLimit-Limit': '5000',
+        }
+        mock_get.return_value = mock_response
+
+        valid, message = validate_token('ghp_valid_token')
+        assert valid is True
+        assert 'testuser' in message
+
+    @patch('gittensor.miner.token_mgmt.requests.get')
+    def test_invalid_token_401(self, mock_get):
+        mock_response = MagicMock()
+        mock_response.status_code = 401
+        mock_get.return_value = mock_response
+
+        valid, message = validate_token('ghp_invalid')
+        assert valid is False
+        assert '401' in message
+
+    @patch('gittensor.miner.token_mgmt.requests.get')
+    def test_rate_limited_403(self, mock_get):
+        mock_response = MagicMock()
+        mock_response.status_code = 403
+        mock_response.headers = {
+            'X-RateLimit-Remaining': '0',
+            'X-RateLimit-Reset': '1700000000',
+        }
+        mock_get.return_value = mock_response
+
+        valid, message = validate_token('ghp_ratelimited')
+        assert valid is False
+        assert 'Rate limited' in message
+
+    @patch('gittensor.miner.token_mgmt.requests.get')
+    @patch('gittensor.miner.token_mgmt.time.sleep')
+    def test_retries_on_timeout(self, mock_sleep, mock_get):
+        import requests as req
+
+        mock_get.side_effect = req.exceptions.Timeout('Connection timed out')
+
+        valid, message = validate_token('ghp_timeout')
+        assert valid is False
+        assert 'Failed after' in message
+        # Should have retried MAX_RETRIES - 1 times (sleep between attempts)
+        assert mock_sleep.call_count == 2  # 3 attempts, 2 sleeps
+
+    @patch('gittensor.miner.token_mgmt.requests.get')
+    @patch('gittensor.miner.token_mgmt.time.sleep')
+    def test_retries_on_connection_error(self, mock_sleep, mock_get):
+        import requests as req
+
+        mock_get.side_effect = req.exceptions.ConnectionError('DNS failure')
+
+        valid, message = validate_token('ghp_connfail')
+        assert valid is False
+        assert mock_sleep.call_count == 2
+
+    @patch('gittensor.miner.token_mgmt.requests.get')
+    @patch('gittensor.miner.token_mgmt.time.sleep')
+    def test_exponential_backoff(self, mock_sleep, mock_get):
+        import requests as req
+
+        mock_get.side_effect = req.exceptions.Timeout('timeout')
+
+        validate_token('ghp_backoff')
+
+        # First sleep: INITIAL_BACKOFF_SECONDS (2.0)
+        # Second sleep: 2.0 * BACKOFF_MULTIPLIER (4.0)
+        calls = mock_sleep.call_args_list
+        assert calls[0][0][0] == pytest.approx(2.0)
+        assert calls[1][0][0] == pytest.approx(4.0)
+
+    @patch('gittensor.miner.token_mgmt.requests.get')
+    def test_does_not_retry_on_401(self, mock_get):
+        """Token revocation (401) should fail immediately without retrying."""
+        mock_response = MagicMock()
+        mock_response.status_code = 401
+        mock_get.return_value = mock_response
+
+        validate_token('ghp_revoked')
+        assert mock_get.call_count == 1
+
+
+class TestIsTokenValid:
+    """Tests for the is_token_valid convenience wrapper."""
+
+    @patch('gittensor.miner.token_mgmt.validate_token')
+    def test_returns_true_for_valid(self, mock_validate):
+        mock_validate.return_value = (True, 'ok')
+        assert is_token_valid('ghp_valid') is True
+
+    @patch('gittensor.miner.token_mgmt.validate_token')
+    def test_returns_false_for_invalid(self, mock_validate):
+        mock_validate.return_value = (False, 'bad')
+        assert is_token_valid('ghp_invalid') is False
+
+
+class TestCheckRateLimit:
+    """Tests for the _check_rate_limit helper."""
+
+    def test_warns_on_low_remaining(self):
+        response = MagicMock()
+        response.headers = {
+            'X-RateLimit-Remaining': '50',
+            'X-RateLimit-Limit': '5000',
+        }
+        # Should not raise
+        _check_rate_limit(response)
+
+    def test_no_warn_on_high_remaining(self):
+        response = MagicMock()
+        response.headers = {
+            'X-RateLimit-Remaining': '4500',
+            'X-RateLimit-Limit': '5000',
+        }
+        _check_rate_limit(response)
+
+    def test_handles_missing_headers(self):
+        response = MagicMock()
+        response.headers = {}
+        _check_rate_limit(response)