Pnant
148 lines
4.8 KiB
Python
148 lines
4.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Tests for Agent Reach CLI."""
|
|
|
|
import json
|
|
import os
|
|
|
|
import pytest
|
|
import requests
|
|
from unittest.mock import patch
|
|
import agent_reach.cli as cli
|
|
from agent_reach.cli import main
|
|
|
|
|
|
class TestCLI:
|
|
def test_version(self, capsys):
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
with patch("sys.argv", ["agent-reach", "version"]):
|
|
main()
|
|
assert exc_info.value.code == 0
|
|
captured = capsys.readouterr()
|
|
assert "Agent Reach v" in captured.out
|
|
|
|
def test_no_command_shows_help(self, capsys):
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
with patch("sys.argv", ["agent-reach"]):
|
|
main()
|
|
assert exc_info.value.code == 0
|
|
|
|
def test_doctor_runs(self, capsys):
|
|
with patch("sys.argv", ["agent-reach", "doctor"]):
|
|
main()
|
|
captured = capsys.readouterr()
|
|
assert "Agent Reach" in captured.out
|
|
assert "✅" in captured.out
|
|
|
|
def test_parse_twitter_cookie_input_separate_values(self):
|
|
auth_token, ct0 = cli._parse_twitter_cookie_input("token123 ct0abc")
|
|
assert auth_token == "token123"
|
|
assert ct0 == "ct0abc"
|
|
|
|
def test_parse_twitter_cookie_input_cookie_header(self):
|
|
auth_token, ct0 = cli._parse_twitter_cookie_input(
|
|
"auth_token=token123; ct0=ct0abc; other=value"
|
|
)
|
|
assert auth_token == "token123"
|
|
assert ct0 == "ct0abc"
|
|
|
|
def test_configure_xhs_cookies_writes_xhs_cli_cookie_file(self, tmp_path, monkeypatch, capsys):
|
|
monkeypatch.setenv("HOME", str(tmp_path))
|
|
monkeypatch.setattr("shutil.which", lambda _name: None)
|
|
|
|
cli._configure_xhs_cookies("a1=token123; web_session=session456; other=value")
|
|
|
|
cookie_path = tmp_path / ".xiaohongshu-cli" / "cookies.json"
|
|
data = json.loads(cookie_path.read_text())
|
|
assert data["a1"] == "token123"
|
|
assert data["web_session"] == "session456"
|
|
assert "saved_at" in data
|
|
assert oct(os.stat(cookie_path).st_mode & 0o777) == "0o600"
|
|
|
|
legacy_path = tmp_path / ".agent-reach" / "xhs-cookies.json"
|
|
assert legacy_path.exists()
|
|
captured = capsys.readouterr()
|
|
assert "xhs-cli cookies saved" 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.4.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
|