fix: split security hardening and setup mcporter config checks
This commit is contained in:
parent
a5682716ec
commit
4b2e6f2ffb
7 changed files with 348 additions and 114 deletions
|
|
@ -2,7 +2,9 @@
|
|||
"""Tests for Agent Reach CLI."""
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
from unittest.mock import patch
|
||||
import agent_reach.cli as cli
|
||||
from agent_reach.cli import main
|
||||
|
||||
|
||||
|
|
@ -27,3 +29,86 @@ class TestCLI:
|
|||
captured = capsys.readouterr()
|
||||
assert "Agent Reach" in captured.out
|
||||
assert "✅" in captured.out
|
||||
|
||||
|
||||
class TestCheckUpdateRetry:
|
||||
def test_retry_timeout_classification(self):
|
||||
sleeps = []
|
||||
|
||||
def fake_sleep(seconds):
|
||||
sleeps.append(seconds)
|
||||
|
||||
with patch("requests.get", side_effect=requests.exceptions.Timeout("timed out")):
|
||||
resp, err, attempts = cli._github_get_with_retry(
|
||||
"https://api.github.com/test",
|
||||
timeout=1,
|
||||
retries=3,
|
||||
sleeper=fake_sleep,
|
||||
)
|
||||
|
||||
assert resp is None
|
||||
assert err == "timeout"
|
||||
assert attempts == 3
|
||||
assert sleeps == [1, 2]
|
||||
|
||||
def test_retry_dns_classification(self):
|
||||
error = requests.exceptions.ConnectionError("getaddrinfo failed for api.github.com")
|
||||
with patch("requests.get", side_effect=error):
|
||||
resp, err, attempts = cli._github_get_with_retry(
|
||||
"https://api.github.com/test",
|
||||
retries=1,
|
||||
sleeper=lambda _x: None,
|
||||
)
|
||||
assert resp is None
|
||||
assert err == "dns"
|
||||
assert attempts == 1
|
||||
|
||||
def test_retry_rate_limit_then_success(self):
|
||||
sleeps = []
|
||||
|
||||
class R:
|
||||
def __init__(self, code, payload=None, headers=None):
|
||||
self.status_code = code
|
||||
self._payload = payload or {}
|
||||
self.headers = headers or {}
|
||||
|
||||
def json(self):
|
||||
return self._payload
|
||||
|
||||
sequence = [
|
||||
R(429, headers={"Retry-After": "3"}),
|
||||
R(200, payload={"tag_name": "v1.2.0"}),
|
||||
]
|
||||
|
||||
with patch("requests.get", side_effect=sequence):
|
||||
resp, err, attempts = cli._github_get_with_retry(
|
||||
"https://api.github.com/test",
|
||||
retries=3,
|
||||
sleeper=lambda s: sleeps.append(s),
|
||||
)
|
||||
|
||||
assert err is None
|
||||
assert resp is not None
|
||||
assert resp.status_code == 200
|
||||
assert attempts == 2
|
||||
assert sleeps == [3.0]
|
||||
|
||||
def test_classify_rate_limit_from_403(self):
|
||||
class R:
|
||||
status_code = 403
|
||||
headers = {"X-RateLimit-Remaining": "0"}
|
||||
|
||||
@staticmethod
|
||||
def json():
|
||||
return {"message": "API rate limit exceeded"}
|
||||
|
||||
assert cli._classify_github_response_error(R()) == "rate_limit"
|
||||
|
||||
def test_check_update_reports_classified_error(self, capsys):
|
||||
with patch("agent_reach.cli._github_get_with_retry", return_value=(None, "timeout", 3)):
|
||||
result = cli._cmd_check_update()
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert result == "error"
|
||||
assert "网络超时" in captured.out
|
||||
assert "已重试 3 次" in captured.out
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue