Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
105 changes: 105 additions & 0 deletions scripts/run_tests.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,105 @@
#!/usr/bin/env python3
"""Run project tests consistently on Windows, Linux, and WSL.

First run in an existing virtual environment:
python scripts/run_tests.py --setup

Subsequent runs:
python scripts/run_tests.py
python scripts/run_tests.py --quick

The --setup option installs this repository's declared [dev] dependencies
into the *active virtual environment*, then runs the tests.
"""
from __future__ import annotations

import argparse
import importlib.util
import os
from pathlib import Path
import subprocess
import sys


REQUIRED_MODULES = ("pytest", "fastapi", "httpx", "uvicorn", "requests")


def run(command: list[str], *, cwd: Path, env: dict[str, str]) -> int:
print("\n$ " + " ".join(command), flush=True)
try:
result = subprocess.run(command, cwd=str(cwd), env=env, check=False)
except OSError as exc:
print(f"Unable to execute command: {exc}", file=sys.stderr)
return 127
print(f"Exit code: {result.returncode}", flush=True)
return result.returncode


def missing_modules() -> list[str]:
return [module for module in REQUIRED_MODULES if importlib.util.find_spec(module) is None]


def main() -> int:
parser = argparse.ArgumentParser(description="Run repository tests with isolated Python dependencies")
parser.add_argument("--quick", action="store_true", help="Run only test_login_headers.py")
parser.add_argument(
"--setup",
action="store_true",
help="Install the project's [dev] dependencies into the active virtual environment before testing",
)
args = parser.parse_args()

root = Path(__file__).resolve().parent.parent
src = root / "src"
if not (src / "codebuddy_proxy").is_dir():
print(f"ERROR: missing package directory: {src / 'codebuddy_proxy'}", file=sys.stderr)
return 2

env = os.environ.copy()
existing_path = env.get("PYTHONPATH", "")
env["PYTHONPATH"] = os.pathsep.join(p for p in (str(src), str(root), existing_path) if p)

print(f"Project: {root}", flush=True)
print(f"Python: {sys.executable}", flush=True)
print(f"PYTHONPATH: {env['PYTHONPATH']}", flush=True)

if args.setup:
if sys.prefix == sys.base_prefix:
print(
"ERROR: --setup requires a virtual environment; refusing to install into system Python.",
file=sys.stderr,
)
print("Create one first: python3 -m venv .venv", file=sys.stderr)
return 2
print("Installing project dependencies, including pytest, from pyproject.toml.", flush=True)
install = run(
[sys.executable, "-m", "pip", "install", "-e", ".[dev]"],
cwd=root,
env=env,
)
if install != 0:
print("RESULT: SETUP FAILED (see pip output above)", file=sys.stderr)
return install

missing = missing_modules()
if missing:
print(f"\nERROR: missing Python modules: {', '.join(missing)}", file=sys.stderr)
print("Within your virtual environment, run this one-time setup:", file=sys.stderr)
print(f' "{sys.executable}" scripts/run_tests.py --setup', file=sys.stderr)
return 2

test_options = [["test_login_headers.py", "-vv", "--tb=long"]]
if not args.quick:
test_options.append(["-q", "--tb=short"])

failures = 0
for options in test_options:
if run([sys.executable, "-m", "pytest", *options], cwd=root, env=env) != 0:
failures += 1

print(f"\nRESULT: {'PASS' if failures == 0 else f'FAIL ({failures} test command(s) failed)'}")
return 0 if failures == 0 else 1


if __name__ == "__main__":
raise SystemExit(main())
9 changes: 8 additions & 1 deletion test_backend_profiles.py
Original file line number Diff line number Diff line change
Expand Up @@ -167,7 +167,14 @@ def test_root_model_config_matches_packaged_domestic_resource():
)
root_copy = Path(__file__).with_name("models_config.json")

assert root_copy.read_bytes() == package_resource.read_bytes()
# Git may check out text files with CRLF on Windows and LF on Linux/WSL.
# Normalize only CRLF vs LF; all other bytes must remain identical.
def normalize_newlines(value: bytes) -> bytes:
return value.replace(b"\r\n", b"\n")

assert normalize_newlines(root_copy.read_bytes()) == normalize_newlines(
package_resource.read_bytes()
)


def test_refresh_preserves_session_metadata_and_machine_id(tmp_path, monkeypatch):
Expand Down
16 changes: 11 additions & 5 deletions test_login_headers.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,14 +4,17 @@
import json
from src.codebuddy_proxy.codebuddy_client_demo import CodeBuddyClient

def test_enterprise_headers():
def test_enterprise_headers(tmp_path):
"""测试企业认证头生成"""
print("=" * 60)
print("测试企业认证头生成")
print("=" * 60)

# 场景1: token 无 domain 字段(最常见的场景)
client = CodeBuddyClient(endpoint="https://copilot.tencent.com")
client = CodeBuddyClient(
endpoint="https://copilot.tencent.com",
session_file=tmp_path / "domestic-session.json",
)
token_no_domain = {"accessToken": "test_token_123"}
headers1 = client._enterprise_headers(token_no_domain)

Expand All @@ -38,7 +41,10 @@ def test_enterprise_headers():
print(" ✓ 正确使用 token.domain 覆盖默认值")

# 场景3: 带端口的 endpoint
client_with_port = CodeBuddyClient(endpoint="https://copilot.example.com:8443")
client_with_port = CodeBuddyClient(
endpoint="https://copilot.example.com:8443",
session_file=tmp_path / "custom-session.json",
)
headers3 = client_with_port._enterprise_headers(token_no_domain)

print("\n场景3: endpoint 包含端口")
Expand All @@ -53,13 +59,13 @@ def test_enterprise_headers():
print("✓ 所有测试通过!")
print("=" * 60)

def test_login_account_headers():
def test_login_account_headers(tmp_path):
"""验证 /login/account 请求会正确组合所有必需的 headers"""
print("\n" + "=" * 60)
print("验证 /login/account 请求头组合")
print("=" * 60)

client = CodeBuddyClient()
client = CodeBuddyClient(session_file=tmp_path / "account-session.json")
mock_token = {"accessToken": "bearer_token_xyz"}

# 模拟 login 方法中的 headers 组合逻辑
Expand Down
74 changes: 49 additions & 25 deletions test_proxy_headers.py
Original file line number Diff line number Diff line change
@@ -1,48 +1,72 @@
#!/usr/bin/env python3
"""测试代理实际发送的headers"""
"""Optional live-proxy integration smoke test.

By default pytest skips this test because it requires a running local proxy,
which may in turn make authenticated upstream requests.

Opt in from a prepared environment:
WORKBUDDY_RUN_PROXY_INTEGRATION=1 python -m pytest test_proxy_headers.py -vv -s

Standalone execution remains available:
python test_proxy_headers.py
"""
from __future__ import annotations

import httpx
import json
import sys
import os

import httpx
import pytest


@pytest.mark.skipif(
os.environ.get("WORKBUDDY_RUN_PROXY_INTEGRATION") != "1",
reason="requires a running proxy; set WORKBUDDY_RUN_PROXY_INTEGRATION=1 to opt in",
)
def test_proxy_headers():
"""测试代理转发时的实际headers"""
proxy_url = "http://localhost:8787/v1/chat/completions"

"""Manually inspect real proxy header forwarding with a running server."""
proxy_url = os.environ.get(
"WORKBUDDY_PROXY_TEST_URL",
"http://127.0.0.1:8787/v1/chat/completions",
)
test_body = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "测试headers"}],
"stream": False,
"max_tokens": 10
"max_tokens": 10,
}

print("=" * 60)
print("测试代理Headers")
print("=" * 60)

try:
resp = httpx.post(
proxy_url,
json=test_body,
headers={"Content-Type": "application/json"},
timeout=30.0
timeout=30.0,
)

print(f"状态码: {resp.status_code}")
print(f"\n响应:")
if resp.status_code == 200:
except httpx.ConnectError as exc:
pytest.fail(
f"无法连接代理服务器 {proxy_url}。请先启动服务,然后重试:{exc}"
)
except httpx.HTTPError as exc:
pytest.fail(f"代理请求失败: {exc}")

print(f"状态码: {resp.status_code}")
if resp.status_code == 200:
try:
print(json.dumps(resp.json(), indent=2, ensure_ascii=False)[:500])
print("\n✅ 请求成功!检查日志确认headers")
else:
print(f"错误: {resp.text[:500]}")

except httpx.ConnectError:
print("❌ 无法连接到代理服务器")
print("请先启动代理: uv run --with workbuddy2api python -m codebuddy_proxy")
sys.exit(1)
except Exception as e:
print(f"❌ 请求失败: {e}")
sys.exit(1)
except ValueError:
print(resp.text[:500])
print("请求成功;请结合服务端日志检查转发的 headers")
else:
print(f"错误: {resp.text[:500]}")
assert resp.status_code == 200, f"代理响应 HTTP {resp.status_code}: {resp.text[:500]}"


if __name__ == "__main__":
# Direct invocation is an explicit opt-in; pytest's skip marker applies
# only when collected through pytest, not when calling this function here.
test_proxy_headers()