File size: 3,552 Bytes
89419ea | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | from src.core.upload import cpa_upload
class FakeResponse:
def __init__(self, status_code=200, payload=None, text=""):
self.status_code = status_code
self._payload = payload
self.text = text
def json(self):
if self._payload is None:
raise ValueError("no json payload")
return self._payload
class FakeMime:
def __init__(self):
self.parts = []
def addpart(self, **kwargs):
self.parts.append(kwargs)
def test_upload_to_cpa_accepts_management_root_url(monkeypatch):
calls = []
def fake_post(url, **kwargs):
calls.append({"url": url, "kwargs": kwargs})
return FakeResponse(status_code=201)
monkeypatch.setattr(cpa_upload, "CurlMime", FakeMime)
monkeypatch.setattr(cpa_upload.cffi_requests, "post", fake_post)
success, message = cpa_upload.upload_to_cpa(
{"email": "tester@example.com"},
api_url="https://cpa.example.com/v0/management",
api_token="token-123",
)
assert success is True
assert message == "上传成功"
assert calls[0]["url"] == "https://cpa.example.com/v0/management/auth-files"
def test_upload_to_cpa_does_not_double_append_full_endpoint(monkeypatch):
calls = []
def fake_post(url, **kwargs):
calls.append({"url": url, "kwargs": kwargs})
return FakeResponse(status_code=201)
monkeypatch.setattr(cpa_upload, "CurlMime", FakeMime)
monkeypatch.setattr(cpa_upload.cffi_requests, "post", fake_post)
success, _ = cpa_upload.upload_to_cpa(
{"email": "tester@example.com"},
api_url="https://cpa.example.com/v0/management/auth-files",
api_token="token-123",
)
assert success is True
assert calls[0]["url"] == "https://cpa.example.com/v0/management/auth-files"
def test_upload_to_cpa_falls_back_to_raw_json_when_multipart_returns_404(monkeypatch):
calls = []
responses = [
FakeResponse(status_code=404, text="404 page not found"),
FakeResponse(status_code=200, payload={"status": "ok"}),
]
def fake_post(url, **kwargs):
calls.append({"url": url, "kwargs": kwargs})
return responses.pop(0)
monkeypatch.setattr(cpa_upload, "CurlMime", FakeMime)
monkeypatch.setattr(cpa_upload.cffi_requests, "post", fake_post)
success, message = cpa_upload.upload_to_cpa(
{"email": "tester@example.com", "type": "codex"},
api_url="https://cpa.example.com",
api_token="token-123",
)
assert success is True
assert message == "上传成功"
assert calls[0]["kwargs"]["multipart"] is not None
assert calls[1]["url"] == "https://cpa.example.com/v0/management/auth-files?name=tester%40example.com.json"
assert calls[1]["kwargs"]["headers"]["Content-Type"] == "application/json"
assert calls[1]["kwargs"]["data"].startswith(b"{")
def test_test_cpa_connection_uses_get_and_normalized_url(monkeypatch):
calls = []
def fake_get(url, **kwargs):
calls.append({"url": url, "kwargs": kwargs})
return FakeResponse(status_code=200, payload={"files": []})
monkeypatch.setattr(cpa_upload.cffi_requests, "get", fake_get)
success, message = cpa_upload.test_cpa_connection(
"https://cpa.example.com/v0/management",
"token-123",
)
assert success is True
assert message == "CPA 连接测试成功"
assert calls[0]["url"] == "https://cpa.example.com/v0/management/auth-files"
assert calls[0]["kwargs"]["headers"]["Authorization"] == "Bearer token-123"
|